This is an automated email from the ASF dual-hosted git repository. jark pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit b5a0b317ad008e031bbfd16e75ebc2a7eb9ae25f Author: JingsongLi <lzljs3620...@aliyun.com> AuthorDate: Thu Aug 29 17:14:21 2019 +0800 [FLINK-13775][table-planner-blink] Refactor ExpressionConverter This closes #9485 --- .../converter/CallExpressionConvertRule.java | 56 ++ .../converter/CustomizedConvertRule.java | 381 +++++++++ .../expressions/converter/DirectConvertRule.java | 157 ++++ .../expressions/converter/ExpressionConverter.java | 907 +++------------------ .../expressions/converter/OverConvertRule.java | 197 +++++ .../converter/ScalarFunctionConvertRule.java | 54 ++ tools/maven/suppressions.xml | 4 +- 7 files changed, 980 insertions(+), 776 deletions(-) diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/CallExpressionConvertRule.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/CallExpressionConvertRule.java new file mode 100644 index 0000000..bf6a5c4 --- /dev/null +++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/CallExpressionConvertRule.java @@ -0,0 +1,56 @@ +/* + * 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.planner.expressions.converter; + +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.planner.calcite.FlinkTypeFactory; + +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.tools.RelBuilder; + +import java.util.Optional; + +/** + * Rule to convert {@link CallExpression}. + */ +public interface CallExpressionConvertRule { + + /** + * Convert call expression with context to RexNode. + * + * @return Success return RexNode of {@link Optional#of}, Fail return {@link Optional#empty()}. + */ + Optional<RexNode> convert(CallExpression call, ConvertContext context); + + /** + * Context of {@link CallExpressionConvertRule}. + */ + interface ConvertContext { + + /** + * Convert expression to RexNode, used by children conversion. + */ + RexNode toRexNode(Expression expr); + + RelBuilder getRelBuilder(); + + FlinkTypeFactory getTypeFactory(); + } +} diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/CustomizedConvertRule.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/CustomizedConvertRule.java new file mode 100644 index 0000000..ca8193b --- /dev/null +++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/CustomizedConvertRule.java @@ -0,0 +1,381 @@ +/* + * 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.planner.expressions.converter; + +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionUtils; +import org.apache.flink.table.expressions.TableReferenceExpression; +import org.apache.flink.table.expressions.TypeLiteralExpression; +import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.expressions.utils.ApiExpressionUtils; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.FunctionDefinition; +import org.apache.flink.table.operations.QueryOperation; +import org.apache.flink.table.planner.calcite.FlinkRelBuilder; +import org.apache.flink.table.planner.functions.InternalFunctionDefinitions; +import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable; +import org.apache.flink.table.planner.functions.sql.SqlThrowExceptionFunction; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import com.google.common.collect.ImmutableList; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexSubQuery; +import org.apache.calcite.sql.fun.SqlTrimFunction; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.apache.calcite.sql.type.SqlTypeName.VARCHAR; +import static org.apache.flink.table.planner.calcite.FlinkTypeFactory.toLogicalType; +import static org.apache.flink.table.planner.expressions.converter.ExpressionConverter.extractValue; +import static org.apache.flink.table.planner.expressions.converter.ExpressionConverter.toRexNodes; +import static org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromDataTypeToLogicalType; +import static org.apache.flink.table.runtime.typeutils.TypeCheckUtils.isCharacterString; +import static org.apache.flink.table.runtime.typeutils.TypeCheckUtils.isTemporal; +import static org.apache.flink.table.runtime.typeutils.TypeCheckUtils.isTimeInterval; + +/** + * Customized {@link CallExpressionConvertRule}, Functions conversion here all require special logic, + * and there may be some special rules, such as needing get the literal values of inputs, such as + * converting to combinations of functions, to convert to RexNode of calcite. + */ +public class CustomizedConvertRule implements CallExpressionConvertRule { + + private static final Map<FunctionDefinition, Conversion> DEFINITION_RULE_MAP = new HashMap<>(); + static { + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.CAST, CustomizedConvertRule::convertCast); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.REINTERPRET_CAST, CustomizedConvertRule::convertReinterpretCast); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.IN, CustomizedConvertRule::convertIn); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.GET, CustomizedConvertRule::convertGet); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.TRIM, CustomizedConvertRule::convertTrim); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.AS, CustomizedConvertRule::convertAs); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.BETWEEN, CustomizedConvertRule::convertBetween); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.NOT_BETWEEN, CustomizedConvertRule::convertNotBetween); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.REPLACE, CustomizedConvertRule::convertReplace); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.PLUS, CustomizedConvertRule::convertPlus); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.CEIL, CustomizedConvertRule::convertCeil); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.FLOOR, CustomizedConvertRule::convertFloor); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.TEMPORAL_OVERLAPS, CustomizedConvertRule::convertTemporalOverlaps); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.TIMESTAMP_DIFF, CustomizedConvertRule::convertTimestampDiff); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.ARRAY, CustomizedConvertRule::convertArray); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.MAP, CustomizedConvertRule::convertMap); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.ROW, CustomizedConvertRule::convertRow); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.ORDER_ASC, CustomizedConvertRule::convertOrderAsc); + DEFINITION_RULE_MAP.put(BuiltInFunctionDefinitions.SQRT, CustomizedConvertRule::convertSqrt); + + // blink expression + DEFINITION_RULE_MAP.put(InternalFunctionDefinitions.THROW_EXCEPTION, CustomizedConvertRule::convertThrowException); + } + + @Override + public Optional<RexNode> convert(CallExpression call, ConvertContext context) { + Conversion conversion = DEFINITION_RULE_MAP.get(call.getFunctionDefinition()); + return Optional.ofNullable(conversion).map(c -> c.convert(call, context)); + } + + private static RexNode convertCast(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 2); + RexNode child = context.toRexNode(call.getChildren().get(0)); + TypeLiteralExpression type = (TypeLiteralExpression) call.getChildren().get(1); + return context.getRelBuilder().getRexBuilder().makeAbstractCast( + context.getTypeFactory().createFieldTypeFromLogicalType( + type.getOutputDataType().getLogicalType().copy(child.getType().isNullable())), + child); + } + + private static RexNode convertOrderAsc(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 1); + return context.toRexNode(call.getChildren().get(0)); + } + + private static RexNode convertTimestampDiff(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 3); + List<RexNode> childrenRexNode = toRexNodes(context, call.getChildren()); + // different orders between flink table api and calcite. + return context.getRelBuilder().call(FlinkSqlOperatorTable.TIMESTAMP_DIFF, childrenRexNode.get(0), childrenRexNode.get(2), + childrenRexNode.get(1)); + } + + private static RexNode convertNotBetween(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 3); + List<RexNode> childrenRexNode = toRexNodes(context, call.getChildren()); + RexNode expr = childrenRexNode.get(0); + RexNode lowerBound = childrenRexNode.get(1); + RexNode upperBound = childrenRexNode.get(2); + return context.getRelBuilder().or( + context.getRelBuilder().call(FlinkSqlOperatorTable.LESS_THAN, expr, lowerBound), + context.getRelBuilder().call(FlinkSqlOperatorTable.GREATER_THAN, expr, upperBound)); + } + + private static RexNode convertBetween(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 3); + List<RexNode> childrenRexNode = toRexNodes(context, call.getChildren()); + RexNode expr = childrenRexNode.get(0); + RexNode lowerBound = childrenRexNode.get(1); + RexNode upperBound = childrenRexNode.get(2); + return context.getRelBuilder().and( + context.getRelBuilder().call(FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL, expr, lowerBound), + context.getRelBuilder().call(FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL, expr, upperBound)); + } + + private static RexNode convertCeil(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 1, 2); + List<Expression> children = call.getChildren(); + List<RexNode> childrenRexNode = toRexNodes(context, children); + if (children.size() == 1) { + return context.getRelBuilder().call(FlinkSqlOperatorTable.CEIL, childrenRexNode); + } else { + return context.getRelBuilder().call(FlinkSqlOperatorTable.CEIL, childrenRexNode.get(1), childrenRexNode.get(0)); + } + } + + private static RexNode convertFloor(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 1, 2); + List<Expression> children = call.getChildren(); + List<RexNode> childrenRexNode = toRexNodes(context, children); + if (children.size() == 1) { + return context.getRelBuilder().call(FlinkSqlOperatorTable.FLOOR, childrenRexNode); + } else { + return context.getRelBuilder().call(FlinkSqlOperatorTable.FLOOR, childrenRexNode.get(1), childrenRexNode.get(0)); + } + } + + private static RexNode convertArray(CallExpression call, ConvertContext context) { + // TODO get type from CallExpression directly until introduce type inference on Expression + List<RexNode> childrenRexNode = toRexNodes(context, call.getChildren()); + ArrayType arrayType = new ArrayType(toLogicalType(childrenRexNode.get(0).getType())); + RelDataType relDataType = context.getTypeFactory().createFieldTypeFromLogicalType(arrayType); + return context.getRelBuilder().getRexBuilder().makeCall(relDataType, FlinkSqlOperatorTable.ARRAY_VALUE_CONSTRUCTOR, childrenRexNode); + } + + private static RexNode convertMap(CallExpression call, ConvertContext context) { + // TODO get type from CallExpression directly until introduce type inference on Expression + List<Expression> children = call.getChildren(); + checkArgument(call, !children.isEmpty() && children.size() % 2 == 0); + List<RexNode> childrenRexNode = toRexNodes(context, children); + RelDataType keyType = childrenRexNode.get(0).getType(); + RelDataType valueType = childrenRexNode.get(childrenRexNode.size() - 1).getType(); + RelDataType mapType = context.getTypeFactory().createMapType(keyType, valueType); + return context.getRelBuilder().getRexBuilder().makeCall(mapType, FlinkSqlOperatorTable.MAP_VALUE_CONSTRUCTOR, childrenRexNode); + } + + private static RexNode convertRow(CallExpression call, ConvertContext context) { + // TODO get type from CallExpression directly until introduce type inference on Expression + List<RexNode> childrenRexNode = toRexNodes(context, call.getChildren()); + LogicalType[] childTypes = childrenRexNode.stream().map(rexNode -> toLogicalType(rexNode.getType())) + .toArray(LogicalType[]::new); + RowType rowType = RowType.of(childTypes); + RelDataType relDataType = context.getTypeFactory().createFieldTypeFromLogicalType(rowType); + return context.getRelBuilder().getRexBuilder().makeCall(relDataType, FlinkSqlOperatorTable.ROW, childrenRexNode); + } + + private static RexNode convertTemporalOverlaps(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 4); + List<RexNode> childrenRexNode = toRexNodes(context, call.getChildren()); + // Standard conversion of the OVERLAPS operator. + // Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#convertOverlaps()]] + RexNode leftTimePoint = childrenRexNode.get(0); + RexNode leftTemporal = childrenRexNode.get(1); + RexNode rightTimePoint = childrenRexNode.get(2); + RexNode rightTemporal = childrenRexNode.get(3); + RexNode convLeftT; + if (isTimeInterval(toLogicalType(leftTemporal.getType()))) { + convLeftT = context.getRelBuilder().call(FlinkSqlOperatorTable.DATETIME_PLUS, leftTimePoint, leftTemporal); + } else { + convLeftT = leftTemporal; + } + // sort end points into start and end, such that (s0 <= e0) and (s1 <= e1). + RexNode leftLe = context.getRelBuilder().call(FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL, leftTimePoint, convLeftT); + RexNode s0 = context.getRelBuilder().call(FlinkSqlOperatorTable.CASE, leftLe, leftTimePoint, convLeftT); + RexNode e0 = context.getRelBuilder().call(FlinkSqlOperatorTable.CASE, leftLe, convLeftT, leftTimePoint); + RexNode convRightT; + if (isTimeInterval(toLogicalType(rightTemporal.getType()))) { + convRightT = context.getRelBuilder().call(FlinkSqlOperatorTable.DATETIME_PLUS, rightTimePoint, rightTemporal); + } else { + convRightT = rightTemporal; + } + RexNode rightLe = context.getRelBuilder().call(FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL, rightTimePoint, convRightT); + RexNode s1 = context.getRelBuilder().call(FlinkSqlOperatorTable.CASE, rightLe, rightTimePoint, convRightT); + RexNode e1 = context.getRelBuilder().call(FlinkSqlOperatorTable.CASE, rightLe, convRightT, rightTimePoint); + + // (e0 >= s1) AND (e1 >= s0) + RexNode leftPred = context.getRelBuilder().call(FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL, e0, s1); + RexNode rightPred = context.getRelBuilder().call(FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL, e1, s0); + return context.getRelBuilder().call(FlinkSqlOperatorTable.AND, leftPred, rightPred); + } + + private static RexNode convertPlus(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 2); + List<RexNode> childrenRexNode = toRexNodes(context, call.getChildren()); + if (isCharacterString(toLogicalType(childrenRexNode.get(0).getType()))) { + return context.getRelBuilder().call( + FlinkSqlOperatorTable.CONCAT, + childrenRexNode.get(0), + context.getRelBuilder().cast(childrenRexNode.get(1), VARCHAR)); + } else if (isCharacterString(toLogicalType(childrenRexNode.get(1).getType()))) { + return context.getRelBuilder().call( + FlinkSqlOperatorTable.CONCAT, + context.getRelBuilder().cast(childrenRexNode.get(0), VARCHAR), + childrenRexNode.get(1)); + } else if (isTimeInterval(toLogicalType(childrenRexNode.get(0).getType())) && + childrenRexNode.get(0).getType() == childrenRexNode.get(1).getType()) { + return context.getRelBuilder().call(FlinkSqlOperatorTable.PLUS, childrenRexNode); + } else if (isTimeInterval(toLogicalType(childrenRexNode.get(0).getType())) + && isTemporal(toLogicalType(childrenRexNode.get(1).getType()))) { + // Calcite has a bug that can't apply INTERVAL + DATETIME (INTERVAL at left) + // we manually switch them here + return context.getRelBuilder().call(FlinkSqlOperatorTable.DATETIME_PLUS, childrenRexNode.get(1), childrenRexNode.get(0)); + } else if (isTemporal(toLogicalType(childrenRexNode.get(0).getType())) && + isTemporal(toLogicalType(childrenRexNode.get(1).getType()))) { + return context.getRelBuilder().call(FlinkSqlOperatorTable.DATETIME_PLUS, childrenRexNode); + } else { + return context.getRelBuilder().call(FlinkSqlOperatorTable.PLUS, childrenRexNode); + } + } + + private static RexNode convertReplace(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 2, 3); + List<Expression> children = call.getChildren(); + List<RexNode> childrenRexNode = toRexNodes(context, children); + if (children.size() == 2) { + return context.getRelBuilder().call( + FlinkSqlOperatorTable.REPLACE, + childrenRexNode.get(0), + childrenRexNode.get(1), + context.getRelBuilder().call(FlinkSqlOperatorTable.CHAR_LENGTH, childrenRexNode.get(0))); + } else { + return context.getRelBuilder().call(FlinkSqlOperatorTable.REPLACE, childrenRexNode); + } + } + + private static RexNode convertAs(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 2); + String name = extractValue((ValueLiteralExpression) call.getChildren().get(1), String.class); + RexNode child = context.toRexNode(call.getChildren().get(0)); + return context.getRelBuilder().alias(child, name); + } + + private static RexNode convertTrim(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 4); + List<Expression> children = call.getChildren(); + ValueLiteralExpression removeLeadingExpr = (ValueLiteralExpression) children.get(0); + Boolean removeLeading = extractValue(removeLeadingExpr, Boolean.class); + ValueLiteralExpression removeTrailingExpr = (ValueLiteralExpression) children.get(1); + Boolean removeTrailing = extractValue(removeTrailingExpr, Boolean.class); + RexNode trimString = context.toRexNode(children.get(2)); + RexNode str = context.toRexNode(children.get(3)); + Enum trimMode; + if (removeLeading && removeTrailing) { + trimMode = SqlTrimFunction.Flag.BOTH; + } else if (removeLeading) { + trimMode = SqlTrimFunction.Flag.LEADING; + } else if (removeTrailing) { + trimMode = SqlTrimFunction.Flag.TRAILING; + } else { + throw new IllegalArgumentException("Unsupported trim mode."); + } + return context.getRelBuilder().call( + FlinkSqlOperatorTable.TRIM, + context.getRelBuilder().getRexBuilder().makeFlag(trimMode), + trimString, + str); + } + + private static RexNode convertGet(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 2); + RexNode child = context.toRexNode(call.getChildren().get(0)); + ValueLiteralExpression keyLiteral = (ValueLiteralExpression) call.getChildren().get(1); + Optional<Integer> indexOptional = ExpressionUtils + .extractValue(keyLiteral, String.class) + .map(child.getType().getFieldNames()::indexOf); + int index = indexOptional.orElseGet(() -> extractValue(keyLiteral, Integer.class)); + return context.getRelBuilder().getRexBuilder().makeFieldAccess(child, index); + } + + private static RexNode convertIn(CallExpression call, ConvertContext context) { + checkArgument(call, call.getChildren().size() > 1); + Expression headExpr = call.getChildren().get(1); + if (headExpr instanceof TableReferenceExpression) { + QueryOperation tableOperation = ((TableReferenceExpression) headExpr).getQueryOperation(); + RexNode child = context.toRexNode(call.getChildren().get(0)); + return RexSubQuery.in(((FlinkRelBuilder) context.getRelBuilder()) + .queryOperation(tableOperation).build(), ImmutableList.of(child)); + } else { + List<RexNode> child = toRexNodes(context, call.getChildren()); + return context.getRelBuilder().call(FlinkSqlOperatorTable.IN, child); + } + } + + private static RexNode convertReinterpretCast(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 3); + RexNode child = context.toRexNode(call.getChildren().get(0)); + TypeLiteralExpression type = (TypeLiteralExpression) call.getChildren().get(1); + RexNode checkOverflow = context.toRexNode(call.getChildren().get(2)); + return context.getRelBuilder().getRexBuilder().makeReinterpretCast( + context.getTypeFactory().createFieldTypeFromLogicalType( + type.getOutputDataType().getLogicalType().copy(child.getType().isNullable())), + child, + checkOverflow); + } + + private static RexNode convertSqrt(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 1); + return context.getRelBuilder().call(FlinkSqlOperatorTable.POWER, toRexNodes(context, + Arrays.asList(call.getChildren().get(0), ApiExpressionUtils.valueLiteral(0.5)))); + } + + private static RexNode convertThrowException(CallExpression call, ConvertContext context) { + checkArgumentNumber(call, 2); + DataType type = ((TypeLiteralExpression) call.getChildren().get(1)).getOutputDataType(); + SqlThrowExceptionFunction function = new SqlThrowExceptionFunction( + context.getTypeFactory().createFieldTypeFromLogicalType(fromDataTypeToLogicalType(type))); + return context.getRelBuilder().call(function, context.toRexNode(call.getChildren().get(0))); + } + + private static void checkArgumentNumber(CallExpression call, int... numbers) { + boolean find = false; + for (int number : numbers) { + if (call.getChildren().size() == number) { + find = true; + break; + } + } + + checkArgument(call, find); + } + + private static void checkArgument(CallExpression call, boolean check) { + if (!check) { + throw new TableException("Invalid arguments for call: " + call); + } + } + + private interface Conversion { + RexNode convert(CallExpression call, ConvertContext context); + } +} diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/DirectConvertRule.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/DirectConvertRule.java new file mode 100644 index 0000000..432b8c7 --- /dev/null +++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/DirectConvertRule.java @@ -0,0 +1,157 @@ +/* + * 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.planner.expressions.converter; + +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.FunctionDefinition; +import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable; + +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.SqlOperator; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import static org.apache.flink.table.planner.expressions.converter.ExpressionConverter.toRexNodes; + +/** + * A {@link CallExpressionConvertRule} that performs a simple one-to-one mapping between + * {@link FunctionDefinition} and a corresponding {@link SqlOperator}. + */ +public class DirectConvertRule implements CallExpressionConvertRule { + + private static final Map<FunctionDefinition, SqlOperator> DEFINITION_OPERATOR_MAP = new HashMap<>(); + static { + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SHA1, FlinkSqlOperatorTable.SHA1); + + // logic functions + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.AND, FlinkSqlOperatorTable.AND); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.OR, FlinkSqlOperatorTable.OR); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.NOT, FlinkSqlOperatorTable.NOT); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.IF, FlinkSqlOperatorTable.CASE); + + // comparison functions + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.EQUALS, FlinkSqlOperatorTable.EQUALS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.GREATER_THAN, FlinkSqlOperatorTable.GREATER_THAN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL, FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LESS_THAN, FlinkSqlOperatorTable.LESS_THAN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL, FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.NOT_EQUALS, FlinkSqlOperatorTable.NOT_EQUALS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.IS_NULL, FlinkSqlOperatorTable.IS_NULL); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.IS_NOT_NULL, FlinkSqlOperatorTable.IS_NOT_NULL); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.IS_TRUE, FlinkSqlOperatorTable.IS_TRUE); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.IS_FALSE, FlinkSqlOperatorTable.IS_FALSE); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.IS_NOT_TRUE, FlinkSqlOperatorTable.IS_NOT_TRUE); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.IS_NOT_FALSE, FlinkSqlOperatorTable.IS_NOT_FALSE); + + // string functions + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.CHAR_LENGTH, FlinkSqlOperatorTable.CHAR_LENGTH); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.INIT_CAP, FlinkSqlOperatorTable.INITCAP); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LIKE, FlinkSqlOperatorTable.LIKE); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LOWER, FlinkSqlOperatorTable.LOWER); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SIMILAR, FlinkSqlOperatorTable.SIMILAR_TO); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SUBSTRING, FlinkSqlOperatorTable.SUBSTRING); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.UPPER, FlinkSqlOperatorTable.UPPER); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.POSITION, FlinkSqlOperatorTable.POSITION); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.OVERLAY, FlinkSqlOperatorTable.OVERLAY); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.CONCAT, FlinkSqlOperatorTable.CONCAT_FUNCTION); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.CONCAT_WS, FlinkSqlOperatorTable.CONCAT_WS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LPAD, FlinkSqlOperatorTable.LPAD); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.RPAD, FlinkSqlOperatorTable.RPAD); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.REGEXP_EXTRACT, FlinkSqlOperatorTable.REGEXP_EXTRACT); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.FROM_BASE64, FlinkSqlOperatorTable.FROM_BASE64); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.TO_BASE64, FlinkSqlOperatorTable.TO_BASE64); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.UUID, FlinkSqlOperatorTable.UUID); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LTRIM, FlinkSqlOperatorTable.LTRIM); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.RTRIM, FlinkSqlOperatorTable.RTRIM); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.REPEAT, FlinkSqlOperatorTable.REPEAT); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.REGEXP_REPLACE, FlinkSqlOperatorTable.REGEXP_REPLACE); + + // math functions + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.MINUS, FlinkSqlOperatorTable.MINUS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.DIVIDE, FlinkSqlOperatorTable.DIVIDE); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.TIMES, FlinkSqlOperatorTable.MULTIPLY); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ABS, FlinkSqlOperatorTable.ABS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.EXP, FlinkSqlOperatorTable.EXP); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LOG10, FlinkSqlOperatorTable.LOG10); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LOG2, FlinkSqlOperatorTable.LOG2); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LN, FlinkSqlOperatorTable.LN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LOG, FlinkSqlOperatorTable.LOG); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.POWER, FlinkSqlOperatorTable.POWER); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.MOD, FlinkSqlOperatorTable.MOD); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.MINUS_PREFIX, FlinkSqlOperatorTable.UNARY_MINUS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SIN, FlinkSqlOperatorTable.SIN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.COS, FlinkSqlOperatorTable.COS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SINH, FlinkSqlOperatorTable.SINH); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.TAN, FlinkSqlOperatorTable.TAN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.TANH, FlinkSqlOperatorTable.TANH); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.COT, FlinkSqlOperatorTable.COT); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ASIN, FlinkSqlOperatorTable.ASIN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ACOS, FlinkSqlOperatorTable.ACOS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ATAN, FlinkSqlOperatorTable.ATAN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ATAN2, FlinkSqlOperatorTable.ATAN2); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.COSH, FlinkSqlOperatorTable.COSH); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.DEGREES, FlinkSqlOperatorTable.DEGREES); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.RADIANS, FlinkSqlOperatorTable.RADIANS); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SIGN, FlinkSqlOperatorTable.SIGN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ROUND, FlinkSqlOperatorTable.ROUND); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.PI, FlinkSqlOperatorTable.PI); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.E, FlinkSqlOperatorTable.E); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.RAND, FlinkSqlOperatorTable.RAND); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.RAND_INTEGER, FlinkSqlOperatorTable.RAND_INTEGER); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.BIN, FlinkSqlOperatorTable.BIN); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.HEX, FlinkSqlOperatorTable.HEX); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.TRUNCATE, FlinkSqlOperatorTable.TRUNCATE); + + // time functions + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.EXTRACT, FlinkSqlOperatorTable.EXTRACT); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.CURRENT_DATE, FlinkSqlOperatorTable.CURRENT_DATE); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.CURRENT_TIME, FlinkSqlOperatorTable.CURRENT_TIME); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.CURRENT_TIMESTAMP, FlinkSqlOperatorTable.CURRENT_TIMESTAMP); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LOCAL_TIME, FlinkSqlOperatorTable.LOCALTIME); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LOCAL_TIMESTAMP, FlinkSqlOperatorTable.LOCALTIMESTAMP); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.DATE_FORMAT, FlinkSqlOperatorTable.DATE_FORMAT); + + // collection + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.AT, FlinkSqlOperatorTable.ITEM); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.CARDINALITY, FlinkSqlOperatorTable.CARDINALITY); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ORDER_DESC, FlinkSqlOperatorTable.DESC); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.ARRAY_ELEMENT, FlinkSqlOperatorTable.ELEMENT); + + // crypto hash + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.MD5, FlinkSqlOperatorTable.MD5); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SHA2, FlinkSqlOperatorTable.SHA2); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SHA224, FlinkSqlOperatorTable.SHA224); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SHA256, FlinkSqlOperatorTable.SHA256); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SHA384, FlinkSqlOperatorTable.SHA384); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SHA512, FlinkSqlOperatorTable.SHA512); + DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.SHA1, FlinkSqlOperatorTable.SHA1); + DEFINITION_OPERATOR_MAP.put( + BuiltInFunctionDefinitions.STREAM_RECORD_TIMESTAMP, FlinkSqlOperatorTable.STREAMRECORD_TIMESTAMP); + } + + @Override + public Optional<RexNode> convert(CallExpression call, ConvertContext context) { + SqlOperator operator = DEFINITION_OPERATOR_MAP.get(call.getFunctionDefinition()); + return Optional.ofNullable(operator) + .map(op -> context.getRelBuilder().call(op, toRexNodes(context, call.getChildren()))); + } +} diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java index b1e8e96..cdd236a 100644 --- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java +++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java @@ -22,71 +22,30 @@ import org.apache.flink.table.api.TableException; import org.apache.flink.table.dataformat.Decimal; import org.apache.flink.table.expressions.CallExpression; import org.apache.flink.table.expressions.Expression; -import org.apache.flink.table.expressions.ExpressionUtils; import org.apache.flink.table.expressions.ExpressionVisitor; import org.apache.flink.table.expressions.FieldReferenceExpression; import org.apache.flink.table.expressions.LocalReferenceExpression; -import org.apache.flink.table.expressions.ResolvedExpression; -import org.apache.flink.table.expressions.TableReferenceExpression; import org.apache.flink.table.expressions.TimeIntervalUnit; import org.apache.flink.table.expressions.TimePointUnit; import org.apache.flink.table.expressions.TypeLiteralExpression; import org.apache.flink.table.expressions.ValueLiteralExpression; -import org.apache.flink.table.expressions.utils.ApiExpressionUtils; -import org.apache.flink.table.functions.AggregateFunction; -import org.apache.flink.table.functions.AggregateFunctionDefinition; -import org.apache.flink.table.functions.BuiltInFunctionDefinitions; -import org.apache.flink.table.functions.FunctionDefinition; -import org.apache.flink.table.functions.ScalarFunction; -import org.apache.flink.table.functions.ScalarFunctionDefinition; -import org.apache.flink.table.functions.TableFunctionDefinition; -import org.apache.flink.table.functions.UserDefinedAggregateFunction; -import org.apache.flink.table.operations.QueryOperation; import org.apache.flink.table.planner.calcite.FlinkContext; -import org.apache.flink.table.planner.calcite.FlinkPlannerImpl; import org.apache.flink.table.planner.calcite.FlinkRelBuilder; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.calcite.RexFieldVariable; import org.apache.flink.table.planner.expressions.RexNodeExpression; -import org.apache.flink.table.planner.expressions.SqlAggFunctionVisitor; -import org.apache.flink.table.planner.functions.InternalFunctionDefinitions; -import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable; -import org.apache.flink.table.planner.functions.sql.SqlThrowExceptionFunction; -import org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils; -import org.apache.flink.table.types.DataType; -import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.planner.expressions.converter.CallExpressionConvertRule.ConvertContext; import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.LogicalType; -import org.apache.flink.table.types.logical.LogicalTypeRoot; -import org.apache.flink.table.types.logical.RowType; -import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; -import org.apache.flink.util.Preconditions; -import com.google.common.collect.ImmutableList; import org.apache.calcite.avatica.util.DateTimeUtils; import org.apache.calcite.avatica.util.TimeUnit; import org.apache.calcite.avatica.util.TimeUnitRange; -import org.apache.calcite.rel.RelFieldCollation; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rex.RexBuilder; -import org.apache.calcite.rex.RexCall; -import org.apache.calcite.rex.RexFieldCollation; import org.apache.calcite.rex.RexNode; -import org.apache.calcite.rex.RexSubQuery; -import org.apache.calcite.rex.RexWindowBound; -import org.apache.calcite.sql.SqlAggFunction; -import org.apache.calcite.sql.SqlBasicCall; -import org.apache.calcite.sql.SqlFunction; import org.apache.calcite.sql.SqlIntervalQualifier; -import org.apache.calcite.sql.SqlKind; -import org.apache.calcite.sql.SqlLiteral; -import org.apache.calcite.sql.SqlNode; -import org.apache.calcite.sql.SqlOperator; -import org.apache.calcite.sql.SqlPostfixOperator; -import org.apache.calcite.sql.SqlWindow; -import org.apache.calcite.sql.fun.SqlTrimFunction; import org.apache.calcite.sql.parser.SqlParserPos; -import org.apache.calcite.sql.type.OrdinalReturnTypeInference; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.calcite.tools.RelBuilder; import org.apache.calcite.util.DateString; @@ -95,572 +54,45 @@ import org.apache.calcite.util.TimestampString; import org.apache.calcite.util.TimestampWithTimeZoneString; import java.math.BigDecimal; -import java.util.ArrayList; import java.util.Arrays; import java.util.Calendar; import java.util.Date; -import java.util.HashSet; -import java.util.IdentityHashMap; import java.util.List; -import java.util.Map; import java.util.Optional; -import java.util.Set; import java.util.TimeZone; import java.util.stream.Collectors; -import static org.apache.calcite.sql.type.SqlTypeName.VARCHAR; -import static org.apache.flink.table.planner.calcite.FlinkTypeFactory.toLogicalType; import static org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromDataTypeToLogicalType; -import static org.apache.flink.table.runtime.typeutils.TypeCheckUtils.isCharacterString; -import static org.apache.flink.table.runtime.typeutils.TypeCheckUtils.isTemporal; -import static org.apache.flink.table.runtime.typeutils.TypeCheckUtils.isTimeInterval; -import static org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType; /** * Visit expression to generator {@link RexNode}. */ public class ExpressionConverter implements ExpressionVisitor<RexNode> { + private static final List<CallExpressionConvertRule> FUNCTION_CONVERT_CHAIN = Arrays.asList( + new ScalarFunctionConvertRule(), + new OverConvertRule(), + new DirectConvertRule(), + new CustomizedConvertRule() + ); + private final RelBuilder relBuilder; private final FlinkTypeFactory typeFactory; - // store mapping from BuiltInFunctionDefinition to it's RexNodeConversion. - private final Map<FunctionDefinition, RexNodeConversion> conversionsOfBuiltInFunc = new IdentityHashMap<>(); - public ExpressionConverter(RelBuilder relBuilder) { this.relBuilder = relBuilder; this.typeFactory = (FlinkTypeFactory) relBuilder.getRexBuilder().getTypeFactory(); - - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.CAST, exprs -> convertCast(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.REINTERPRET_CAST, exprs -> convertReinterpretCast(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.IN, exprs -> convertIn(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.GET, exprs -> convertGet(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.TRIM, exprs -> convertTrim(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.AS, exprs -> convertAs(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.OVER, exprs -> convertOver(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.IS_NULL, exprs -> convertIsNull(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.BETWEEN, exprs -> convertBetween(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.NOT_BETWEEN, exprs -> convertNotBetween(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.REPLACE, exprs -> convertReplace(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.PLUS, exprs -> convertPlus(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.CEIL, exprs -> convertCeil(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.FLOOR, exprs -> convertFloor(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.TEMPORAL_OVERLAPS, exprs -> convertTemporalOverlaps(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.TIMESTAMP_DIFF, exprs -> convertTimestampDiff(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ARRAY, exprs -> convertArray(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ARRAY_ELEMENT, exprs -> convertArrayElement(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.MAP, exprs -> convertMap(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ROW, exprs -> convertRow(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ORDER_ASC, exprs -> convertOrderAsc(exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.SHA1, exprs -> convert(FlinkSqlOperatorTable.SHA1, exprs)); - // logic functions - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.AND, exprs -> convert(FlinkSqlOperatorTable.AND, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.OR, exprs -> convert(FlinkSqlOperatorTable.OR, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.NOT, exprs -> convert(FlinkSqlOperatorTable.NOT, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.IF, exprs -> convert(FlinkSqlOperatorTable.CASE, exprs)); - - // comparison functions - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.EQUALS, exprs -> convert(FlinkSqlOperatorTable.EQUALS, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.GREATER_THAN, exprs -> convert(FlinkSqlOperatorTable.GREATER_THAN, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL, exprs -> convert(FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.LESS_THAN, exprs -> convert(FlinkSqlOperatorTable.LESS_THAN, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL, exprs -> convert(FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.NOT_EQUALS, exprs -> convert(FlinkSqlOperatorTable.NOT_EQUALS, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.IS_NULL, exprs -> convert(FlinkSqlOperatorTable.IS_NULL, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.IS_NOT_NULL, exprs -> convert(FlinkSqlOperatorTable.IS_NOT_NULL, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.IS_TRUE, exprs -> convert(FlinkSqlOperatorTable.IS_TRUE, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.IS_FALSE, exprs -> convert(FlinkSqlOperatorTable.IS_FALSE, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.IS_NOT_TRUE, exprs -> convert(FlinkSqlOperatorTable.IS_NOT_TRUE, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.IS_NOT_FALSE, exprs -> convert(FlinkSqlOperatorTable.IS_NOT_FALSE, exprs)); - - // string functions - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.CHAR_LENGTH, exprs -> convert(FlinkSqlOperatorTable.CHAR_LENGTH, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.INIT_CAP, exprs -> convert(FlinkSqlOperatorTable.INITCAP, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.LIKE, exprs -> convert(FlinkSqlOperatorTable.LIKE, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.LOWER, exprs -> convert(FlinkSqlOperatorTable.LOWER, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.SIMILAR, exprs -> convert(FlinkSqlOperatorTable.SIMILAR_TO, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.SUBSTRING, exprs -> convert(FlinkSqlOperatorTable.SUBSTRING, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.UPPER, exprs -> convert(FlinkSqlOperatorTable.UPPER, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.POSITION, exprs -> convert(FlinkSqlOperatorTable.POSITION, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.OVERLAY, exprs -> convert(FlinkSqlOperatorTable.OVERLAY, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.CONCAT, exprs -> convert(FlinkSqlOperatorTable.CONCAT_FUNCTION, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.CONCAT_WS, exprs -> convert(FlinkSqlOperatorTable.CONCAT_WS, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.LPAD, exprs -> convert(FlinkSqlOperatorTable.LPAD, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.RPAD, exprs -> convert(FlinkSqlOperatorTable.RPAD, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.REGEXP_EXTRACT, exprs -> convert(FlinkSqlOperatorTable.REGEXP_EXTRACT, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.FROM_BASE64, exprs -> convert(FlinkSqlOperatorTable.FROM_BASE64, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.TO_BASE64, exprs -> convert(FlinkSqlOperatorTable.TO_BASE64, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.UUID, exprs -> convert(FlinkSqlOperatorTable.UUID, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.LTRIM, exprs -> convert(FlinkSqlOperatorTable.LTRIM, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.RTRIM, exprs -> convert(FlinkSqlOperatorTable.RTRIM, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.REPEAT, exprs -> convert(FlinkSqlOperatorTable.REPEAT, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.REGEXP_REPLACE, exprs -> convert(FlinkSqlOperatorTable.REGEXP_REPLACE, exprs)); - - // math functions - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.MINUS, exprs -> convert(FlinkSqlOperatorTable.MINUS, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.DIVIDE, exprs -> convert(FlinkSqlOperatorTable.DIVIDE, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.TIMES, exprs -> convert(FlinkSqlOperatorTable.MULTIPLY, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ABS, exprs -> convert(FlinkSqlOperatorTable.ABS, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.EXP, exprs -> convert(FlinkSqlOperatorTable.EXP, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.LOG10, exprs -> convert(FlinkSqlOperatorTable.LOG10, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.LOG2, exprs -> convert(FlinkSqlOperatorTable.LOG2, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.LN, exprs -> convert(FlinkSqlOperatorTable.LN, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.LOG, exprs -> convert(FlinkSqlOperatorTable.LOG, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.POWER, exprs -> convert(FlinkSqlOperatorTable.POWER, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.MOD, exprs -> convert(FlinkSqlOperatorTable.MOD, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.SQRT, exprs -> - relBuilder.call(FlinkSqlOperatorTable.POWER, - convertCallChildren(Arrays.asList(exprs.get(0), ApiExpressionUtils.valueLiteral(0.5))))); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.MINUS_PREFIX, exprs -> convert(FlinkSqlOperatorTable.UNARY_MINUS, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.SIN, exprs -> convert(FlinkSqlOperatorTable.SIN, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.COS, exprs -> convert(FlinkSqlOperatorTable.COS, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.SINH, exprs -> convert(FlinkSqlOperatorTable.SINH, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.TAN, exprs -> convert(FlinkSqlOperatorTable.TAN, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.TANH, exprs -> convert(FlinkSqlOperatorTable.TANH, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.COT, exprs -> convert(FlinkSqlOperatorTable.COT, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ASIN, exprs -> convert(FlinkSqlOperatorTable.ASIN, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ACOS, exprs -> convert(FlinkSqlOperatorTable.ACOS, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.ATAN, exprs -> convert(FlinkSqlOperatorTable.ATAN, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.ATAN2, exprs -> convert(FlinkSqlOperatorTable.ATAN2, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.COSH, exprs -> convert(FlinkSqlOperatorTable.COSH, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.DEGREES, exprs -> convert(FlinkSqlOperatorTable.DEGREES, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.RADIANS, exprs -> convert(FlinkSqlOperatorTable.RADIANS, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.SIGN, exprs -> convert(FlinkSqlOperatorTable.SIGN, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.ROUND, exprs -> convert(FlinkSqlOperatorTable.ROUND, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.PI, exprs -> convert(FlinkSqlOperatorTable.PI, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.E, exprs -> convert(FlinkSqlOperatorTable.E, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.RAND, exprs -> convert(FlinkSqlOperatorTable.RAND, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.RAND_INTEGER, exprs -> convert(FlinkSqlOperatorTable.RAND_INTEGER, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.BIN, exprs -> convert(FlinkSqlOperatorTable.BIN, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.HEX, exprs -> convert(FlinkSqlOperatorTable.HEX, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.TRUNCATE, exprs -> convert(FlinkSqlOperatorTable.TRUNCATE, exprs)); - - // time functions - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.EXTRACT, exprs -> convert(FlinkSqlOperatorTable.EXTRACT, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.CURRENT_DATE, exprs -> convert(FlinkSqlOperatorTable.CURRENT_DATE, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.CURRENT_TIME, exprs -> convert(FlinkSqlOperatorTable.CURRENT_TIME, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.CURRENT_TIMESTAMP, exprs -> convert(FlinkSqlOperatorTable.CURRENT_TIMESTAMP, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.LOCAL_TIME, exprs -> convert(FlinkSqlOperatorTable.LOCALTIME, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.LOCAL_TIMESTAMP, exprs -> convert(FlinkSqlOperatorTable.LOCALTIMESTAMP, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.DATE_FORMAT, exprs -> convert(FlinkSqlOperatorTable.DATE_FORMAT, exprs)); - - // collection - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.AT, exprs -> convert(FlinkSqlOperatorTable.ITEM, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.CARDINALITY, exprs -> convert(FlinkSqlOperatorTable.CARDINALITY, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.ORDER_DESC, exprs -> convert(FlinkSqlOperatorTable.DESC, exprs)); - - // crypto hash - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.MD5, exprs -> convert(FlinkSqlOperatorTable.MD5, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.SHA2, exprs -> convert(FlinkSqlOperatorTable.SHA2, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.SHA224, exprs -> convert(FlinkSqlOperatorTable.SHA224, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.SHA256, exprs -> convert(FlinkSqlOperatorTable.SHA256, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.SHA384, exprs -> convert(FlinkSqlOperatorTable.SHA384, exprs)); - conversionsOfBuiltInFunc - .put(BuiltInFunctionDefinitions.SHA512, exprs -> convert(FlinkSqlOperatorTable.SHA512, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.SHA1, exprs -> convert(FlinkSqlOperatorTable.SHA1, exprs)); - conversionsOfBuiltInFunc.put(BuiltInFunctionDefinitions.STREAM_RECORD_TIMESTAMP, - exprs -> convert(FlinkSqlOperatorTable.STREAMRECORD_TIMESTAMP, exprs)); - - // blink expression - conversionsOfBuiltInFunc.put(InternalFunctionDefinitions.THROW_EXCEPTION, exprs -> { - DataType type = ((TypeLiteralExpression) exprs.get(1)).getOutputDataType(); - return convert(new SqlThrowExceptionFunction( - typeFactory.createFieldTypeFromLogicalType(fromDataTypeToLogicalType(type))), - exprs.subList(0, 1)); - }); } @Override public RexNode visit(CallExpression call) { - FunctionDefinition func = call.getFunctionDefinition(); - if (func instanceof ScalarFunctionDefinition) { - ScalarFunction scalaFunc = ((ScalarFunctionDefinition) func).getScalarFunction(); - List<RexNode> child = convertCallChildren(call.getChildren()); - SqlFunction sqlFunction = UserDefinedFunctionUtils.createScalarSqlFunction( - scalaFunc.functionIdentifier(), - scalaFunc.toString(), - scalaFunc, - typeFactory); - return relBuilder.call(sqlFunction, child); - } else if (func instanceof TableFunctionDefinition) { - throw new RuntimeException("There is no possible reach here!"); - } else if (func instanceof AggregateFunctionDefinition) { - UserDefinedAggregateFunction aggFunc = ((AggregateFunctionDefinition) func).getAggregateFunction(); - if (aggFunc instanceof AggregateFunction) { - SqlFunction aggSqlFunction = UserDefinedFunctionUtils.createAggregateSqlFunction( - aggFunc.functionIdentifier(), - aggFunc.toString(), - (AggregateFunction) aggFunc, - fromLegacyInfoToDataType(aggFunc.getResultType()), - fromLegacyInfoToDataType(aggFunc.getAccumulatorType()), - typeFactory); - List<RexNode> child = convertCallChildren(call.getChildren()); - return relBuilder.call(aggSqlFunction, child); - } else { - throw new UnsupportedOperationException("TableAggregateFunction is not supported yet!"); + for (CallExpressionConvertRule rule : FUNCTION_CONVERT_CHAIN) { + Optional<RexNode> converted = rule.convert(call, newFunctionContext()); + if (converted.isPresent()) { + return converted.get(); } - - } else { - FunctionDefinition def = call.getFunctionDefinition(); - if (conversionsOfBuiltInFunc.containsKey(def)) { - RexNodeConversion conversion = conversionsOfBuiltInFunc.get(def); - return conversion.convert(call.getChildren()); - } else { - throw new UnsupportedOperationException(def.toString()); - } - } - } - - private List<RexNode> convertCallChildren(List<Expression> children) { - return children.stream() - .map(expression -> expression.accept(ExpressionConverter.this)) - .collect(Collectors.toList()); - } - - private RexNode convert(SqlOperator sqlOperator, List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - return relBuilder.call(sqlOperator, childrenRexNode); - } - - private RexNode convertArrayElement(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - return relBuilder.call(FlinkSqlOperatorTable.ELEMENT, childrenRexNode); - } - - private RexNode convertOrderAsc(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - return childrenRexNode.get(0); - } - - private RexNode convertTimestampDiff(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - return relBuilder.call(FlinkSqlOperatorTable.TIMESTAMP_DIFF, childrenRexNode.get(0), childrenRexNode.get(2), - childrenRexNode.get(1)); - } - - private RexNode convertIsNull(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - return relBuilder.isNull(childrenRexNode.get(0)); - } - - private RexNode convertNotBetween(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - Preconditions.checkArgument(childrenRexNode.size() == 3); - RexNode expr = childrenRexNode.get(0); - RexNode lowerBound = childrenRexNode.get(1); - RexNode upperBound = childrenRexNode.get(2); - return relBuilder.or( - relBuilder.call(FlinkSqlOperatorTable.LESS_THAN, expr, lowerBound), - relBuilder.call(FlinkSqlOperatorTable.GREATER_THAN, expr, upperBound)); - } - - private RexNode convertBetween(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - Preconditions.checkArgument(childrenRexNode.size() == 3); - RexNode expr = childrenRexNode.get(0); - RexNode lowerBound = childrenRexNode.get(1); - RexNode upperBound = childrenRexNode.get(2); - return relBuilder.and( - relBuilder.call(FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL, expr, lowerBound), - relBuilder.call(FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL, expr, upperBound)); - } - - private RexNode convertCeil(List<Expression> children) { - Preconditions.checkArgument(children.size() == 1 || children.size() == 2); - List<RexNode> childrenRexNode = convertCallChildren(children); - if (children.size() == 1) { - return relBuilder.call(FlinkSqlOperatorTable.CEIL, childrenRexNode); - } else { - return relBuilder.call(FlinkSqlOperatorTable.CEIL, childrenRexNode.get(1), childrenRexNode.get(0)); - } - } - - private RexNode convertFloor(List<Expression> children) { - Preconditions.checkArgument(children.size() == 1 || children.size() == 2); - List<RexNode> childrenRexNode = convertCallChildren(children); - if (children.size() == 1) { - return relBuilder.call(FlinkSqlOperatorTable.FLOOR, childrenRexNode); - } else { - return relBuilder.call(FlinkSqlOperatorTable.FLOOR, childrenRexNode.get(1), childrenRexNode.get(0)); - } - } - - private RexNode convertArray(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - ArrayType arrayType = new ArrayType(toLogicalType(childrenRexNode.get(0).getType())); - // TODO get type from CallExpression directly - RelDataType relDataType = typeFactory.createFieldTypeFromLogicalType(arrayType); - return relBuilder.getRexBuilder().makeCall(relDataType, FlinkSqlOperatorTable.ARRAY_VALUE_CONSTRUCTOR, childrenRexNode); - } - - private RexNode convertMap(List<Expression> children) { - Preconditions.checkArgument(!children.isEmpty() && children.size() % 2 == 0); - // TODO get type from CallExpression directly - List<RexNode> childrenRexNode = convertCallChildren(children); - RelDataType keyType = childrenRexNode.get(0).getType(); - RelDataType valueType = childrenRexNode.get(childrenRexNode.size() - 1).getType(); - RelDataType mapType = typeFactory.createMapType(keyType, valueType); - return relBuilder.getRexBuilder().makeCall(mapType, FlinkSqlOperatorTable.MAP_VALUE_CONSTRUCTOR, childrenRexNode); - } - - private RexNode convertRow(List<Expression> children) { - // TODO get type from CallExpression directly - List<RexNode> childrenRexNode = convertCallChildren(children); - LogicalType[] childTypes = childrenRexNode.stream().map(rexNode -> toLogicalType(rexNode.getType())) - .toArray(LogicalType[]::new); - RowType rowType = RowType.of(childTypes); - RelDataType relDataType = typeFactory.createFieldTypeFromLogicalType(rowType); - return relBuilder.getRexBuilder().makeCall(relDataType, FlinkSqlOperatorTable.ROW, childrenRexNode); - } - - private RexNode convertTemporalOverlaps(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - // Standard conversion of the OVERLAPS operator. - // Source: [[org.apache.calcite.sql2rel.StandardConvertletTable#convertOverlaps()]] - RexNode leftTimePoint = childrenRexNode.get(0); - RexNode leftTemporal = childrenRexNode.get(1); - RexNode rightTimePoint = childrenRexNode.get(2); - RexNode rightTemporal = childrenRexNode.get(3); - RexNode convLeftT; - if (isTimeInterval(toLogicalType(leftTemporal.getType()))) { - convLeftT = relBuilder.call(FlinkSqlOperatorTable.DATETIME_PLUS, leftTimePoint, leftTemporal); - } else { - convLeftT = leftTemporal; } - // sort end points into start and end, such that (s0 <= e0) and (s1 <= e1). - RexNode leftLe = relBuilder.call(FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL, leftTimePoint, convLeftT); - RexNode s0 = relBuilder.call(FlinkSqlOperatorTable.CASE, leftLe, leftTimePoint, convLeftT); - RexNode e0 = relBuilder.call(FlinkSqlOperatorTable.CASE, leftLe, convLeftT, leftTimePoint); - RexNode convRightT; - if (isTimeInterval(toLogicalType(rightTemporal.getType()))) { - convRightT = relBuilder.call(FlinkSqlOperatorTable.DATETIME_PLUS, rightTimePoint, rightTemporal); - } else { - convRightT = rightTemporal; - } - RexNode rightLe = relBuilder.call(FlinkSqlOperatorTable.LESS_THAN_OR_EQUAL, rightTimePoint, convRightT); - RexNode s1 = relBuilder.call(FlinkSqlOperatorTable.CASE, rightLe, rightTimePoint, convRightT); - RexNode e1 = relBuilder.call(FlinkSqlOperatorTable.CASE, rightLe, convRightT, rightTimePoint); - - // (e0 >= s1) AND (e1 >= s0) - RexNode leftPred = relBuilder.call(FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL, e0, s1); - RexNode rightPred = relBuilder.call(FlinkSqlOperatorTable.GREATER_THAN_OR_EQUAL, e1, s0); - return relBuilder.call(FlinkSqlOperatorTable.AND, leftPred, rightPred); - } - - private RexNode convertPlus(List<Expression> children) { - List<RexNode> childrenRexNode = convertCallChildren(children); - if (isCharacterString(toLogicalType(childrenRexNode.get(0).getType()))) { - return relBuilder.call( - FlinkSqlOperatorTable.CONCAT, - childrenRexNode.get(0), - relBuilder.cast(childrenRexNode.get(1), VARCHAR)); - } else if (isCharacterString(toLogicalType(childrenRexNode.get(1).getType()))) { - return relBuilder.call( - FlinkSqlOperatorTable.CONCAT, - relBuilder.cast(childrenRexNode.get(0), VARCHAR), - childrenRexNode.get(1)); - } else if (isTimeInterval(toLogicalType(childrenRexNode.get(0).getType())) && - childrenRexNode.get(0).getType() == childrenRexNode.get(1).getType()) { - return relBuilder.call(FlinkSqlOperatorTable.PLUS, childrenRexNode); - } else if (isTimeInterval(toLogicalType(childrenRexNode.get(0).getType())) - && isTemporal(toLogicalType(childrenRexNode.get(1).getType()))) { - // Calcite has a bug that can't apply INTERVAL + DATETIME (INTERVAL at left) - // we manually switch them here - return relBuilder.call(FlinkSqlOperatorTable.DATETIME_PLUS, childrenRexNode.get(1), childrenRexNode.get(0)); - } else if (isTemporal(toLogicalType(childrenRexNode.get(0).getType())) && - isTemporal(toLogicalType(childrenRexNode.get(1).getType()))) { - return relBuilder.call(FlinkSqlOperatorTable.DATETIME_PLUS, childrenRexNode); - } else { - return relBuilder.call(FlinkSqlOperatorTable.PLUS, childrenRexNode); - } - } - - private RexNode convertReplace(List<Expression> children) { - Preconditions.checkArgument(children.size() == 2 || children.size() == 3); - List<RexNode> childrenRexNode = convertCallChildren(children); - if (children.size() == 2) { - return relBuilder.call( - FlinkSqlOperatorTable.REPLACE, - childrenRexNode.get(0), - childrenRexNode.get(1), - relBuilder.call(FlinkSqlOperatorTable.CHAR_LENGTH, childrenRexNode.get(0))); - } else { - return relBuilder.call(FlinkSqlOperatorTable.REPLACE, childrenRexNode); - } - } - - private RexNode convertOver(List<Expression> children) { - List<Expression> args = children; - Expression agg = args.get(0); - SqlAggFunction aggFunc = agg.accept(new SqlAggFunctionVisitor(typeFactory)); - RelDataType aggResultType = typeFactory.createFieldTypeFromLogicalType( - fromDataTypeToLogicalType(((ResolvedExpression) agg).getOutputDataType())); - - // assemble exprs by agg children - List<RexNode> aggExprs = agg.getChildren().stream().map(expr -> expr.accept(this)) - .collect(Collectors.toList()); - - // assemble order by key - Expression orderKeyExpr = args.get(1); - Set<SqlKind> kinds = new HashSet<>(); - RexNode collationRexNode = createCollation(orderKeyExpr.accept(this), RelFieldCollation.Direction.ASCENDING, - null, kinds); - ImmutableList<RexFieldCollation> orderKey = ImmutableList - .of(new RexFieldCollation(collationRexNode, kinds)); - - // assemble partition by keys - List<RexNode> partitionKeys = args.subList(4, args.size()).stream().map(expr -> expr.accept(this)) - .collect(Collectors.toList()); - // assemble bounds - Expression preceding = args.get(2); - boolean isPhysical = LogicalTypeChecks.hasRoot( - fromDataTypeToLogicalType(((ResolvedExpression) preceding).getOutputDataType()), - LogicalTypeRoot.BIGINT); - Expression following = args.get(3); - RexWindowBound lowerBound = createBound(preceding, SqlKind.PRECEDING); - RexWindowBound upperBound = createBound(following, SqlKind.FOLLOWING); - - // build RexOver - return relBuilder.getRexBuilder().makeOver( - aggResultType, - aggFunc, - aggExprs, - partitionKeys, - orderKey, - lowerBound, - upperBound, - isPhysical, - true, - false, - false); - } - - private RexNode convertAs(List<Expression> children) { - String name = extractValue((ValueLiteralExpression) children.get(1), String.class); - RexNode child = children.get(0).accept(this); - return relBuilder.alias(child, name); - } - - private RexNode convertTrim(List<Expression> children) { - ValueLiteralExpression removeLeadingExpr = (ValueLiteralExpression) children.get(0); - Boolean removeLeading = extractValue(removeLeadingExpr, Boolean.class); - ValueLiteralExpression removeTrailingExpr = (ValueLiteralExpression) children.get(1); - Boolean removeTrailing = extractValue(removeTrailingExpr, Boolean.class); - RexNode trimString = children.get(2).accept(this); - RexNode str = children.get(3).accept(this); - Enum trimMode; - if (removeLeading && removeTrailing) { - trimMode = SqlTrimFunction.Flag.BOTH; - } else if (removeLeading) { - trimMode = SqlTrimFunction.Flag.LEADING; - } else if (removeTrailing) { - trimMode = SqlTrimFunction.Flag.TRAILING; - } else { - throw new IllegalArgumentException("Unsupported trim mode."); - } - return relBuilder.call( - FlinkSqlOperatorTable.TRIM, - relBuilder.getRexBuilder().makeFlag(trimMode), - trimString, - str); - } - - private RexNode convertGet(List<Expression> children) { - RexNode child = children.get(0).accept(this); - ValueLiteralExpression keyLiteral = (ValueLiteralExpression) children.get(1); - Optional<Integer> indexOptional = ExpressionUtils.extractValue(keyLiteral, String.class).map( - child.getType().getFieldNames()::indexOf); - // Note: never replace the following code with : - // int index = indexOptional.orElseGet(() -> extractValue(keyLiteral, Integer.class)); - // Because the logical in `orElseGet` always executed no matter whether indexOptional is present or not. - int index; - if (indexOptional.isPresent()) { - index = indexOptional.get(); - } else { - index = extractValue(keyLiteral, Integer.class); - } - return relBuilder.getRexBuilder().makeFieldAccess(child, index); - } - - private RexNode convertIn(List<Expression> children) { - Expression headExpr = children.get(1); - if (headExpr instanceof TableReferenceExpression) { - QueryOperation tableOperation = ((TableReferenceExpression) headExpr).getQueryOperation(); - RexNode child = children.get(0).accept(this); - return RexSubQuery.in( - ((FlinkRelBuilder) relBuilder).queryOperation(tableOperation).build(), - ImmutableList.of(child)); - } else { - List<RexNode> child = convertCallChildren(children); - return relBuilder.call(FlinkSqlOperatorTable.IN, child); - } - } - - private RexNode convertReinterpretCast(List<Expression> children) { - RexNode child = children.get(0).accept(this); - TypeLiteralExpression type = (TypeLiteralExpression) children.get(1); - RexNode checkOverflow = children.get(2).accept(this); - return relBuilder.getRexBuilder().makeReinterpretCast( - typeFactory.createFieldTypeFromLogicalType( - type.getOutputDataType().getLogicalType().copy(child.getType().isNullable())), - child, - checkOverflow); - } - - private RexNode convertCast(List<Expression> children) { - RexNode child = children.get(0).accept(this); - TypeLiteralExpression type = (TypeLiteralExpression) children.get(1); - return relBuilder.getRexBuilder().makeAbstractCast( - typeFactory.createFieldTypeFromLogicalType( - type.getOutputDataType().getLogicalType().copy(child.getType().isNullable())), - child); + throw new RuntimeException("Unknown call expression: " + call); } @Override @@ -728,140 +160,23 @@ public class ExpressionConverter implements ExpressionVisitor<RexNode> { } Object object = extractValue(valueLiteral, Object.class); if (object instanceof TimePointUnit) { - TimeUnit value; - switch ((TimePointUnit) object) { - case YEAR: - value = TimeUnit.YEAR; - break; - case MONTH: - value = TimeUnit.MONTH; - break; - case DAY: - value = TimeUnit.DAY; - break; - case HOUR: - value = TimeUnit.HOUR; - break; - case MINUTE: - value = TimeUnit.MINUTE; - break; - case SECOND: - value = TimeUnit.SECOND; - break; - case QUARTER: - value = TimeUnit.QUARTER; - break; - case WEEK: - value = TimeUnit.WEEK; - break; - case MILLISECOND: - value = TimeUnit.MILLISECOND; - break; - case MICROSECOND: - value = TimeUnit.MICROSECOND; - break; - default: - throw new UnsupportedOperationException(); - } + TimeUnit value = timePointUnitToTimeUnit((TimePointUnit) object); return relBuilder.getRexBuilder().makeFlag(value); } else if (object instanceof TimeIntervalUnit) { - TimeUnitRange value; - switch ((TimeIntervalUnit) object) { - case YEAR: - value = TimeUnitRange.YEAR; - break; - case YEAR_TO_MONTH: - value = TimeUnitRange.YEAR_TO_MONTH; - break; - case QUARTER: - value = TimeUnitRange.QUARTER; - break; - case MONTH: - value = TimeUnitRange.MONTH; - break; - case WEEK: - value = TimeUnitRange.WEEK; - break; - case DAY: - value = TimeUnitRange.DAY; - break; - case DAY_TO_HOUR: - value = TimeUnitRange.DAY_TO_HOUR; - break; - case DAY_TO_MINUTE: - value = TimeUnitRange.DAY_TO_MINUTE; - break; - case DAY_TO_SECOND: - value = TimeUnitRange.DAY_TO_SECOND; - break; - case HOUR: - value = TimeUnitRange.HOUR; - break; - case SECOND: - value = TimeUnitRange.SECOND; - break; - case HOUR_TO_MINUTE: - value = TimeUnitRange.HOUR_TO_MINUTE; - break; - case HOUR_TO_SECOND: - value = TimeUnitRange.HOUR_TO_SECOND; - break; - case MINUTE: - value = TimeUnitRange.MINUTE; - break; - case MINUTE_TO_SECOND: - value = TimeUnitRange.MINUTE_TO_SECOND; - break; - default: - throw new UnsupportedOperationException(); - } + TimeUnitRange value = intervalUnitToUnitRange((TimeIntervalUnit) object); return relBuilder.getRexBuilder().makeFlag(value); } else { return relBuilder.literal(extractValue(valueLiteral, Object.class)); } } - /** - * Extracts a value from a literal. Including planner-specific instances such as {@link Decimal}. - */ - @SuppressWarnings("unchecked") - private static <T> T extractValue(ValueLiteralExpression literal, Class<T> clazz) { - final Optional<Object> possibleObject = literal.getValueAs(Object.class); - if (!possibleObject.isPresent()) { - throw new TableException("Invalid literal."); - } - final Object object = possibleObject.get(); - - if (clazz.equals(BigDecimal.class)) { - final Optional<BigDecimal> possibleDecimal = literal.getValueAs(BigDecimal.class); - if (possibleDecimal.isPresent()) { - return (T) possibleDecimal.get(); - } - if (object instanceof Decimal) { - return (T) ((Decimal) object).toBigDecimal(); - } - } - - return literal.getValueAs(clazz) - .orElseThrow(() -> new TableException("Unsupported literal class: " + clazz)); - } - - /** - * Convert a Date value to a Calendar. Calcite's fromCalendarField functions use the - * Calendar.get methods, so the raw values of the individual fields are preserved when - * converted to the String formats. - * - * @return get the Calendar value - */ - private static Calendar valueAsCalendar(Object value) { - Date date = (Date) value; - Calendar cal = Calendar.getInstance(); - cal.setTime(date); - return cal; - } - @Override public RexNode visit(FieldReferenceExpression fieldReference) { + // We can not use inputCount+inputIndex+FieldIndex to construct field of calcite. + // See QueryOperationConverter.SingleRelVisitor.visit(AggregateQueryOperation). + // Calcite will shuffle the output order of groupings. + // So the output fields order will be changed too. + // See RelBuilder.aggregate, it use ImmutableBitSet to store groupings, return relBuilder.field(fieldReference.getName()); } @@ -885,85 +200,129 @@ public class ExpressionConverter implements ExpressionVisitor<RexNode> { } } - private RexNode createCollation(RexNode node, RelFieldCollation.Direction direction, - RelFieldCollation.NullDirection nullDirection, Set<SqlKind> kinds) { - switch (node.getKind()) { - case DESCENDING: - kinds.add(node.getKind()); - return createCollation(((RexCall) node).getOperands().get(0), RelFieldCollation.Direction.DESCENDING, - nullDirection, kinds); - case NULLS_FIRST: - kinds.add(node.getKind()); - return createCollation(((RexCall) node).getOperands().get(0), direction, - RelFieldCollation.NullDirection.FIRST, kinds); - case NULLS_LAST: - kinds.add(node.getKind()); - return createCollation(((RexCall) node).getOperands().get(0), direction, - RelFieldCollation.NullDirection.LAST, kinds); + public static List<RexNode> toRexNodes(ConvertContext context, List<Expression> expr) { + return expr.stream().map(context::toRexNode).collect(Collectors.toList()); + } + + private ConvertContext newFunctionContext() { + return new ConvertContext() { + @Override + public RexNode toRexNode(Expression expr) { + return expr.accept(ExpressionConverter.this); + } + + @Override + public RelBuilder getRelBuilder() { + return relBuilder; + } + + @Override + public FlinkTypeFactory getTypeFactory() { + return typeFactory; + } + }; + } + + private static TimeUnit timePointUnitToTimeUnit(TimePointUnit unit) { + switch (unit) { + case YEAR: + return TimeUnit.YEAR; + case MONTH: + return TimeUnit.MONTH; + case DAY: + return TimeUnit.DAY; + case HOUR: + return TimeUnit.HOUR; + case MINUTE: + return TimeUnit.MINUTE; + case SECOND: + return TimeUnit.SECOND; + case QUARTER: + return TimeUnit.QUARTER; + case WEEK: + return TimeUnit.WEEK; + case MILLISECOND: + return TimeUnit.MILLISECOND; + case MICROSECOND: + return TimeUnit.MICROSECOND; default: - if (nullDirection == null) { - // Set the null direction if not specified. - // Consistent with HIVE/SPARK/MYSQL/BLINK-RUNTIME. - if (FlinkPlannerImpl.defaultNullCollation() - .last(direction.equals(RelFieldCollation.Direction.DESCENDING))) { - kinds.add(SqlKind.NULLS_LAST); - } else { - kinds.add(SqlKind.NULLS_FIRST); - } - } - return node; + throw new UnsupportedOperationException("TimePointUnit is: " + unit); } } - private RexWindowBound createBound(Expression bound, SqlKind sqlKind) { - if (bound instanceof CallExpression) { - CallExpression callExpr = (CallExpression) bound; - FunctionDefinition func = callExpr.getFunctionDefinition(); - if (BuiltInFunctionDefinitions.UNBOUNDED_ROW.equals(func) || BuiltInFunctionDefinitions.UNBOUNDED_RANGE - .equals(func)) { - SqlNode unbounded = sqlKind.equals(SqlKind.PRECEDING) ? SqlWindow - .createUnboundedPreceding(SqlParserPos.ZERO) : - SqlWindow.createUnboundedFollowing(SqlParserPos.ZERO); - return RexWindowBound.create(unbounded, null); - } else if (BuiltInFunctionDefinitions.CURRENT_ROW.equals(func) || BuiltInFunctionDefinitions.CURRENT_RANGE - .equals(func)) { - SqlNode currentRow = SqlWindow.createCurrentRow(SqlParserPos.ZERO); - return RexWindowBound.create(currentRow, null); - } else { - throw new IllegalArgumentException("Unexpected expression: " + bound); - } - } else if (bound instanceof ValueLiteralExpression) { - RelDataType returnType = typeFactory - .createFieldTypeFromLogicalType(new DecimalType(true, 19, 0)); - SqlOperator sqlOperator = new SqlPostfixOperator( - sqlKind.name(), - sqlKind, - 2, - new OrdinalReturnTypeInference(0), - null, - null); - SqlNode[] operands = new SqlNode[] { SqlLiteral.createExactNumeric("1", SqlParserPos.ZERO) }; - SqlNode node = new SqlBasicCall(sqlOperator, operands, SqlParserPos.ZERO); + private static TimeUnitRange intervalUnitToUnitRange(TimeIntervalUnit intervalUnit) { + switch (intervalUnit) { + case YEAR: + return TimeUnitRange.YEAR; + case YEAR_TO_MONTH: + return TimeUnitRange.YEAR_TO_MONTH; + case QUARTER: + return TimeUnitRange.QUARTER; + case MONTH: + return TimeUnitRange.MONTH; + case WEEK: + return TimeUnitRange.WEEK; + case DAY: + return TimeUnitRange.DAY; + case DAY_TO_HOUR: + return TimeUnitRange.DAY_TO_HOUR; + case DAY_TO_MINUTE: + return TimeUnitRange.DAY_TO_MINUTE; + case DAY_TO_SECOND: + return TimeUnitRange.DAY_TO_SECOND; + case HOUR: + return TimeUnitRange.HOUR; + case SECOND: + return TimeUnitRange.SECOND; + case HOUR_TO_MINUTE: + return TimeUnitRange.HOUR_TO_MINUTE; + case HOUR_TO_SECOND: + return TimeUnitRange.HOUR_TO_SECOND; + case MINUTE: + return TimeUnitRange.MINUTE; + case MINUTE_TO_SECOND: + return TimeUnitRange.MINUTE_TO_SECOND; + default: + throw new UnsupportedOperationException("TimeIntervalUnit is: " + intervalUnit); + } + } - ValueLiteralExpression literalExpr = (ValueLiteralExpression) bound; - RexNode literalRexNode = literalExpr.getValueAs(Double.class).map( - v -> relBuilder.literal(BigDecimal.valueOf((Double) v))).orElse( - relBuilder.literal(extractValue(literalExpr, Object.class))); + /** + * Extracts a value from a literal. Including planner-specific instances such as {@link Decimal}. + */ + @SuppressWarnings("unchecked") + public static <T> T extractValue(ValueLiteralExpression literal, Class<T> clazz) { + final Optional<Object> possibleObject = literal.getValueAs(Object.class); + if (!possibleObject.isPresent()) { + throw new TableException("Invalid literal."); + } + final Object object = possibleObject.get(); - List<RexNode> expressions = new ArrayList<>(); - expressions.add(literalRexNode); - RexNode rexNode = relBuilder.getRexBuilder().makeCall(returnType, sqlOperator, expressions); - return RexWindowBound.create(node, rexNode); - } else { - throw new TableException("Unexpected expression: " + bound); + if (clazz.equals(BigDecimal.class)) { + final Optional<BigDecimal> possibleDecimal = literal.getValueAs(BigDecimal.class); + if (possibleDecimal.isPresent()) { + return (T) possibleDecimal.get(); + } + if (object instanceof Decimal) { + return (T) ((Decimal) object).toBigDecimal(); + } } + + return literal.getValueAs(clazz) + .orElseThrow(() -> new TableException("Unsupported literal class: " + clazz)); } /** - * RexNodeConversion to define how to convert a {@link CallExpression} which - * has built-in FunctionDefinition to RexNode. + * Convert a Date value to a Calendar. Calcite's fromCalendarField functions use the + * Calendar.get methods, so the raw values of the individual fields are preserved when + * converted to the String formats. + * + * @return get the Calendar value */ - private interface RexNodeConversion { - RexNode convert(List<Expression> children); + private static Calendar valueAsCalendar(Object value) { + Date date = (Date) value; + Calendar cal = Calendar.getInstance(); + cal.setTime(date); + return cal; } } diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java new file mode 100644 index 0000000..7d8173c --- /dev/null +++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java @@ -0,0 +1,197 @@ +/* + * 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.planner.expressions.converter; + +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.FunctionDefinition; +import org.apache.flink.table.planner.calcite.FlinkPlannerImpl; +import org.apache.flink.table.planner.calcite.FlinkTypeFactory; +import org.apache.flink.table.planner.expressions.SqlAggFunctionVisitor; +import org.apache.flink.table.types.logical.DecimalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; + +import com.google.common.collect.ImmutableList; +import org.apache.calcite.rel.RelFieldCollation; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexFieldCollation; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexWindowBound; +import org.apache.calcite.sql.SqlAggFunction; +import org.apache.calcite.sql.SqlBasicCall; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlLiteral; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlOperator; +import org.apache.calcite.sql.SqlPostfixOperator; +import org.apache.calcite.sql.SqlWindow; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.sql.type.OrdinalReturnTypeInference; + +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.flink.table.planner.expressions.converter.ExpressionConverter.extractValue; +import static org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromDataTypeToLogicalType; + +/** + * A {@link CallExpressionConvertRule} that converts {@link BuiltInFunctionDefinitions#OVER}. + */ +public class OverConvertRule implements CallExpressionConvertRule { + + @Override + public Optional<RexNode> convert(CallExpression call, ConvertContext context) { + List<Expression> children = call.getChildren(); + if (call.getFunctionDefinition() == BuiltInFunctionDefinitions.OVER) { + FlinkTypeFactory typeFactory = context.getTypeFactory(); + Expression agg = children.get(0); + SqlAggFunction aggFunc = agg.accept(new SqlAggFunctionVisitor(typeFactory)); + RelDataType aggResultType = typeFactory.createFieldTypeFromLogicalType( + fromDataTypeToLogicalType(((ResolvedExpression) agg).getOutputDataType())); + + // assemble exprs by agg children + List<RexNode> aggExprs = agg.getChildren().stream().map(context::toRexNode) + .collect(Collectors.toList()); + + // assemble order by key + Expression orderKeyExpr = children.get(1); + Set<SqlKind> kinds = new HashSet<>(); + RexNode collationRexNode = createCollation( + context.toRexNode(orderKeyExpr), RelFieldCollation.Direction.ASCENDING, null, kinds); + ImmutableList<RexFieldCollation> orderKey = ImmutableList + .of(new RexFieldCollation(collationRexNode, kinds)); + + // assemble partition by keys + List<RexNode> partitionKeys = children.subList(4, children.size()).stream().map(context::toRexNode) + .collect(Collectors.toList()); + // assemble bounds + Expression preceding = children.get(2); + boolean isPhysical = LogicalTypeChecks.hasRoot( + fromDataTypeToLogicalType(((ResolvedExpression) preceding).getOutputDataType()), + LogicalTypeRoot.BIGINT); + Expression following = children.get(3); + RexWindowBound lowerBound = createBound(context, preceding, SqlKind.PRECEDING); + RexWindowBound upperBound = createBound(context, following, SqlKind.FOLLOWING); + + // build RexOver + return Optional.of(context.getRelBuilder().getRexBuilder().makeOver( + aggResultType, + aggFunc, + aggExprs, + partitionKeys, + orderKey, + lowerBound, + upperBound, + isPhysical, + true, + false, + false)); + } + return Optional.empty(); + } + + private RexNode createCollation( + RexNode node, + RelFieldCollation.Direction direction, + RelFieldCollation.NullDirection nullDirection, + Set<SqlKind> kinds) { + switch (node.getKind()) { + case DESCENDING: + kinds.add(node.getKind()); + return createCollation(((RexCall) node).getOperands().get(0), RelFieldCollation.Direction.DESCENDING, + nullDirection, kinds); + case NULLS_FIRST: + kinds.add(node.getKind()); + return createCollation(((RexCall) node).getOperands().get(0), direction, + RelFieldCollation.NullDirection.FIRST, kinds); + case NULLS_LAST: + kinds.add(node.getKind()); + return createCollation(((RexCall) node).getOperands().get(0), direction, + RelFieldCollation.NullDirection.LAST, kinds); + default: + if (nullDirection == null) { + // Set the null direction if not specified. + // Consistent with HIVE/SPARK/MYSQL/BLINK-RUNTIME. + if (FlinkPlannerImpl.defaultNullCollation() + .last(direction.equals(RelFieldCollation.Direction.DESCENDING))) { + kinds.add(SqlKind.NULLS_LAST); + } else { + kinds.add(SqlKind.NULLS_FIRST); + } + } + return node; + } + } + + private RexWindowBound createBound(ConvertContext context, Expression bound, SqlKind sqlKind) { + if (bound instanceof CallExpression) { + CallExpression callExpr = (CallExpression) bound; + FunctionDefinition func = callExpr.getFunctionDefinition(); + if (BuiltInFunctionDefinitions.UNBOUNDED_ROW.equals(func) || BuiltInFunctionDefinitions.UNBOUNDED_RANGE + .equals(func)) { + SqlNode unbounded = sqlKind.equals(SqlKind.PRECEDING) ? SqlWindow + .createUnboundedPreceding(SqlParserPos.ZERO) : + SqlWindow.createUnboundedFollowing(SqlParserPos.ZERO); + return RexWindowBound.create(unbounded, null); + } else if (BuiltInFunctionDefinitions.CURRENT_ROW.equals(func) || BuiltInFunctionDefinitions.CURRENT_RANGE + .equals(func)) { + SqlNode currentRow = SqlWindow.createCurrentRow(SqlParserPos.ZERO); + return RexWindowBound.create(currentRow, null); + } else { + throw new IllegalArgumentException("Unexpected expression: " + bound); + } + } else if (bound instanceof ValueLiteralExpression) { + RelDataType returnType = context.getTypeFactory() + .createFieldTypeFromLogicalType(new DecimalType(true, 19, 0)); + SqlOperator sqlOperator = new SqlPostfixOperator( + sqlKind.name(), + sqlKind, + 2, + new OrdinalReturnTypeInference(0), + null, + null); + SqlNode[] operands = new SqlNode[] { SqlLiteral.createExactNumeric("1", SqlParserPos.ZERO) }; + SqlNode node = new SqlBasicCall(sqlOperator, operands, SqlParserPos.ZERO); + + ValueLiteralExpression literalExpr = (ValueLiteralExpression) bound; + RexNode literalRexNode = literalExpr.getValueAs(Double.class).map( + v -> context.getRelBuilder().literal(BigDecimal.valueOf((Double) v))).orElse( + context.getRelBuilder().literal(extractValue(literalExpr, Object.class))); + + List<RexNode> expressions = new ArrayList<>(); + expressions.add(literalRexNode); + RexNode rexNode = context.getRelBuilder().getRexBuilder().makeCall( + returnType, sqlOperator, expressions); + return RexWindowBound.create(node, rexNode); + } else { + throw new TableException("Unexpected expression: " + bound); + } + } +} diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/ScalarFunctionConvertRule.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/ScalarFunctionConvertRule.java new file mode 100644 index 0000000..8921092 --- /dev/null +++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/converter/ScalarFunctionConvertRule.java @@ -0,0 +1,54 @@ +/* + * 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.planner.expressions.converter; + +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.functions.FunctionDefinition; +import org.apache.flink.table.functions.ScalarFunction; +import org.apache.flink.table.functions.ScalarFunctionDefinition; +import org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils; + +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.SqlFunction; + +import java.util.Optional; + +import static org.apache.flink.table.planner.expressions.converter.ExpressionConverter.toRexNodes; + +/** + * {@link CallExpressionConvertRule} to convert {@link ScalarFunctionDefinition}. + */ +public class ScalarFunctionConvertRule implements CallExpressionConvertRule { + + @Override + public Optional<RexNode> convert(CallExpression call, ConvertContext context) { + FunctionDefinition def = call.getFunctionDefinition(); + if (def instanceof ScalarFunctionDefinition) { + ScalarFunction scalaFunc = ((ScalarFunctionDefinition) def).getScalarFunction(); + SqlFunction sqlFunction = UserDefinedFunctionUtils.createScalarSqlFunction( + scalaFunc.functionIdentifier(), + scalaFunc.toString(), + scalaFunc, + context.getTypeFactory()); + return Optional.of(context.getRelBuilder() + .call(sqlFunction, toRexNodes(context, call.getChildren()))); + } + return Optional.empty(); + } +} diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml index 7635914..4a56880 100644 --- a/tools/maven/suppressions.xml +++ b/tools/maven/suppressions.xml @@ -35,9 +35,9 @@ under the License. <suppress files="org[\\/]apache[\\/]flink[\\/]streaming[\\/]python[\\/]connectors[\\/].*.java" checks="MethodNameCheck|ParameterName"/> - <!-- RexNodeConverter has to use guava directly --> + <!-- Have to use guava directly --> <suppress - files="RexNodeConverter.java" + files="OverConvertRule.java|CustomizedConvertRule.java" checks="IllegalImport"/> <!-- Cassandra connectors have to use guava directly --> <suppress