twalthr commented on a change in pull request #11290: [FLINK-16379][table] Introduce fromValues in TableEnvironment URL: https://github.com/apache/flink/pull/11290#discussion_r404107297
########## File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/ValuesOperationFactory.java ########## @@ -0,0 +1,201 @@ +/* + * 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.operations.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionDefaultVisitor; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.expressions.resolver.ExpressionResolver; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.operations.QueryOperation; +import org.apache.flink.table.operations.ValuesQueryOperation; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.utils.LogicalTypeGeneralization; +import org.apache.flink.table.types.utils.TypeConversions; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral; + +/** + * Utility class for creating valid {@link ValuesQueryOperation} operation. + */ +@Internal +class ValuesOperationFactory { + /** + * Creates a valid {@link ValuesQueryOperation} operation. + * + * <p>It derives a row type based on {@link LogicalTypeGeneralization}. It flattens any + * row constructors. It does not flatten ROWs which are a result of e.g. a function call. + * + * <p>The resulting schema can be provided manually. If it is not, the schema will be automatically derived from + * the types of the expressions. + */ + QueryOperation create( + @Nullable TableSchema expectedSchema, + List<ResolvedExpression> resolvedExpressions, + ExpressionResolver.PostResolverFactory postResolverFactory) { + List<List<ResolvedExpression>> resolvedRows = unwrapFromRowConstructor(resolvedExpressions); + + TableSchema schema = Optional.ofNullable(expectedSchema) + .orElseGet(() -> extractSchema(resolvedRows)); + + List<List<ResolvedExpression>> castedExpressions = resolvedRows.stream() + .map(row -> convertToExpectedRowType(postResolverFactory, schema.getFieldDataTypes(), row)) + .collect(Collectors.toList()); + + return new ValuesQueryOperation(castedExpressions, schema); + } + + private TableSchema extractSchema(List<List<ResolvedExpression>> resolvedRows) { + DataType[] dataTypes = findRowType(resolvedRows); + String[] fieldNames = IntStream.range(0, dataTypes.length) + .mapToObj(i -> "f" + i) + .toArray(String[]::new); + return TableSchema.builder() + .fields(fieldNames, dataTypes) + .build(); + } + + private List<ResolvedExpression> convertToExpectedRowType( + ExpressionResolver.PostResolverFactory postResolverFactory, + DataType[] dataTypes, + List<ResolvedExpression> row) { + return IntStream.range(0, row.size()) + .mapToObj(i -> { + boolean typesMatch = row.get(i) + .getOutputDataType() + .getLogicalType() + .equals(dataTypes[i].getLogicalType()); + if (typesMatch) { + return row.get(i); + } + + ResolvedExpression castedExpr = row.get(i); + DataType targetType = dataTypes[i]; + return convertToExpectedType(castedExpr, targetType, postResolverFactory); + }) + .collect(Collectors.toList()); + } + + private ResolvedExpression convertToExpectedType( + ResolvedExpression castedExpr, + DataType targetType, + ExpressionResolver.PostResolverFactory postResolverFactory) { + + // if the expression is a literal try converting the literal in place instead of casting + if (castedExpr instanceof ValueLiteralExpression) { + Optional<?> convertedValue = ((ValueLiteralExpression) castedExpr).getValueAs(targetType.getConversionClass()); + if (convertedValue.isPresent()) { + return valueLiteral(convertedValue.get(), targetType); + } + } + return postResolverFactory.cast(castedExpr, targetType); + } + + private List<List<ResolvedExpression>> unwrapFromRowConstructor(List<ResolvedExpression> resolvedExpressions) { + return resolvedExpressions + .stream() + .map(expr -> expr.accept( + new ExpressionDefaultVisitor<List<ResolvedExpression>>() { + @Override + public List<ResolvedExpression> visit(CallExpression call) { + if (call.getFunctionDefinition() == BuiltInFunctionDefinitions.ROW) { + return call.getResolvedChildren(); + } + + return defaultMethod(call); + } + + @Override + protected List<ResolvedExpression> defaultMethod(Expression expression) { + if (!(expression instanceof ResolvedExpression)) { + throw new TableException( + "This visitor is applied to ResolvedExpressions. We should never end up here."); + } + + return Collections.singletonList((ResolvedExpression) expression); + } + })) + .collect(Collectors.toList()); + } + + private DataType[] findRowType(List<List<ResolvedExpression>> resolvedRows) { + int rowSize = findRowSize(resolvedRows); + DataType[] dataTypes = new DataType[rowSize]; + IntStream.range(0, rowSize).forEach(i -> { + dataTypes[i] = findCommonTypeAtPosition(resolvedRows, i); + }); + return dataTypes; + } + + private DataType findCommonTypeAtPosition(List<List<ResolvedExpression>> resolvedRows, int i) { + List<LogicalType> typesAtIPosition = extractLogicalTypesAtPosition(resolvedRows, i); + + LogicalType logicalType = LogicalTypeGeneralization.findCommonType(typesAtIPosition) + .orElseThrow(() -> new ValidationException(String.format( + "Types in VALUES must match. Could not find a common type at a %d-th position.", + i))); + + return TypeConversions.fromLogicalToDataType(logicalType); + } + + private List<LogicalType> extractLogicalTypesAtPosition( + List<List<ResolvedExpression>> resolvedRows, + int rowPosition) { + List<LogicalType> typesAtIPosition = new ArrayList<>(); + for (List<ResolvedExpression> resolvedExpression : resolvedRows) { + LogicalType outputDataType = resolvedExpression.get(rowPosition).getOutputDataType().getLogicalType(); + typesAtIPosition.add(outputDataType); + } + return typesAtIPosition; + } + + private int findRowSize(List<List<ResolvedExpression>> resolvedRows) { + List<ResolvedExpression> firstRow = resolvedRows.get(0); + int potentialRowSize = firstRow.size(); + verifyAllSameSize(resolvedRows, potentialRowSize); + return potentialRowSize; + } + + private void verifyAllSameSize(List<List<ResolvedExpression>> resolvedRows, int potentialRowSize) { + Optional<List<ResolvedExpression>> differentSizeRow = resolvedRows.stream() + .filter(row -> row.size() != potentialRowSize) + .findAny(); + if (differentSizeRow.isPresent()) { + throw new ValidationException(String.format( + "All rows in VALUES clause must have the same arity. Row %s have different arity than the first row.", Review comment: `All rows in a fromValues(...) clause must have the same field length. Row %s has a different length than the first row.` arity is very Flink specific. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
