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_r404118834
########## 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); Review comment: Should we add an additional check if the expression can be casted to the target type? The post resolver factory should be called with care because no validation is performed anymore when the schema is manually defined. ---------------------------------------------------------------- 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
