twalthr commented on a change in pull request #8860: [FLINK-12906][table-planner] Ported OperationTreeBuilder to java URL: https://github.com/apache/flink/pull/8860#discussion_r299445656
########## File path: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/operations/OperationTreeBuilderImpl.java ########## @@ -0,0 +1,699 @@ +/* + * 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; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.table.api.GroupWindow; +import org.apache.flink.table.api.OverWindow; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.api.Types; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.FunctionLookup; +import org.apache.flink.table.expressions.ApiExpressionDefaultVisitor; +import org.apache.flink.table.expressions.ApiExpressionUtils; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.ExpressionBridge; +import org.apache.flink.table.expressions.ExpressionResolver; +import org.apache.flink.table.expressions.ExpressionUtils; +import org.apache.flink.table.expressions.LocalReferenceExpression; +import org.apache.flink.table.expressions.LookupCallResolver; +import org.apache.flink.table.expressions.PlannerExpression; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.expressions.UnresolvedCallExpression; +import org.apache.flink.table.expressions.UnresolvedReferenceExpression; +import org.apache.flink.table.expressions.lookups.TableReferenceLookup; +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.FunctionKind; +import org.apache.flink.table.functions.TableFunctionDefinition; +import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils; +import org.apache.flink.table.operations.JoinQueryOperation.JoinType; +import org.apache.flink.table.operations.WindowAggregateQueryOperation.ResolvedGroupWindow; +import org.apache.flink.table.typeutils.FieldInfoUtils; +import org.apache.flink.util.Preconditions; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +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.expressions.ApiExpressionUtils.unresolvedCall; +import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef; +import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral; +import static org.apache.flink.table.operations.SetQueryOperation.SetQueryOperationType.INTERSECT; +import static org.apache.flink.table.operations.SetQueryOperation.SetQueryOperationType.MINUS; +import static org.apache.flink.table.operations.SetQueryOperation.SetQueryOperationType.UNION; + +/** + * A builder for constructing validated {@link QueryOperation}s. + */ +@Internal +public class OperationTreeBuilderImpl implements OperationTreeBuilder { + + private final ExpressionBridge<PlannerExpression> expressionBridge; + private final FunctionLookup functionCatalog; + private final TableReferenceLookup tableReferenceLookup; + private final LookupCallResolver lookupCallResolver; + + /** + * Utility classes for constructing a validated operation of certain type. + */ + private final ProjectionOperationFactory projectionOperationFactory; + private final SortOperationFactory sortOperationFactory; + private final CalculatedTableFactory calculatedTableFactory; + private final SetOperationFactory setOperationFactory; + private final AggregateOperationFactory aggregateOperationFactory; + private final JoinOperationFactory joinOperationFactory; + + private OperationTreeBuilderImpl( + ExpressionBridge<PlannerExpression> expressionBridge, + FunctionLookup functionLookup, + TableReferenceLookup tableReferenceLookup, + ProjectionOperationFactory projectionOperationFactory, + SortOperationFactory sortOperationFactory, + CalculatedTableFactory calculatedTableFactory, + SetOperationFactory setOperationFactory, + AggregateOperationFactory aggregateOperationFactory, + JoinOperationFactory joinOperationFactory) { + this.expressionBridge = expressionBridge; + this.functionCatalog = functionLookup; + this.tableReferenceLookup = tableReferenceLookup; + this.projectionOperationFactory = projectionOperationFactory; + this.sortOperationFactory = sortOperationFactory; + this.calculatedTableFactory = calculatedTableFactory; + this.setOperationFactory = setOperationFactory; + this.aggregateOperationFactory = aggregateOperationFactory; + this.joinOperationFactory = joinOperationFactory; + this.lookupCallResolver = new LookupCallResolver(functionLookup); + } + + public static OperationTreeBuilderImpl create( + ExpressionBridge<PlannerExpression> expressionBridge, + FunctionLookup functionCatalog, + TableReferenceLookup tableReferenceLookup, + boolean isStreaming) { + return new OperationTreeBuilderImpl( + expressionBridge, + functionCatalog, + tableReferenceLookup, + new ProjectionOperationFactory(expressionBridge), + new SortOperationFactory(isStreaming), + new CalculatedTableFactory(), + new SetOperationFactory(isStreaming), + new AggregateOperationFactory(expressionBridge, isStreaming), + new JoinOperationFactory(expressionBridge) + ); + } + + public QueryOperation project(List<Expression> projectList, QueryOperation child) { + return project(projectList, child, false); + } + + public QueryOperation project(List<Expression> projectList, QueryOperation child, boolean explicitAlias) { + projectList.forEach(p -> p.accept(new NoAggregateChecker( + "Aggregate functions are not supported in the select right after the aggregate" + + " or flatAggregate operation."))); + projectList.forEach(p -> p.accept(new NoWindowPropertyChecker( + "Window properties can only be used on windowed tables."))); + return projectInternal(projectList, child, explicitAlias, Collections.emptyList()); + } + + public QueryOperation project(List<Expression> projectList, QueryOperation child, List<OverWindow> overWindows) { + + Preconditions.checkArgument(!overWindows.isEmpty()); + + projectList.forEach(p -> p.accept(new NoWindowPropertyChecker( + "Window start and end properties are not available for Over windows."))); + + return projectInternal( + projectList, + child, + true, + overWindows); + } + + private QueryOperation projectInternal( + List<Expression> projectList, + QueryOperation child, + boolean explicitAlias, + List<OverWindow> overWindows) { + + ExpressionResolver resolver = ExpressionResolver.resolverFor(tableReferenceLookup, functionCatalog, child) + .withOverWindows(overWindows) + .build(); + List<ResolvedExpression> projections = resolver.resolve(projectList); + return projectionOperationFactory.create(projections, child, explicitAlias, resolver.postResolverFactory()); + } + + /** + * Adds additional columns. Existing fields will be replaced if replaceIfExist is true. + */ + public QueryOperation addColumns(boolean replaceIfExist, List<Expression> fieldLists, QueryOperation child) { + final List<Expression> newColumns; + if (replaceIfExist) { + String[] fieldNames = child.getTableSchema().getFieldNames(); + newColumns = ColumnOperationUtils.addOrReplaceColumns(Arrays.asList(fieldNames), fieldLists); + } else { + newColumns = new ArrayList<>(fieldLists); + newColumns.add(0, new UnresolvedReferenceExpression("*")); + } + return project(newColumns, child, false); + } + + public QueryOperation renameColumns(List<Expression> aliases, QueryOperation child) { + + ExpressionResolver resolver = getResolver(child); + String[] inputFieldNames = child.getTableSchema().getFieldNames(); + List<Expression> validateAliases = ColumnOperationUtils.renameColumns( + Arrays.asList(inputFieldNames), + resolver.resolveExpanding(aliases)); + + return project(validateAliases, child, false); + } + + public QueryOperation dropColumns(List<Expression> fieldLists, QueryOperation child) { + + ExpressionResolver resolver = getResolver(child); + String[] inputFieldNames = child.getTableSchema().getFieldNames(); + List<Expression> finalFields = ColumnOperationUtils.dropFields( + Arrays.asList(inputFieldNames), + resolver.resolveExpanding(fieldLists)); + + return project(finalFields, child, false); + } + + public QueryOperation aggregate(List<Expression> groupingExpressions, List<Expression> aggregates, QueryOperation child) { + + ExpressionResolver resolver = getResolver(child); + + List<ResolvedExpression> resolvedGroupings = resolver.resolve(groupingExpressions); + List<ResolvedExpression> resolvedAggregates = resolver.resolve(aggregates); + + return aggregateOperationFactory.createAggregate(resolvedGroupings, resolvedAggregates, child); + } + + public QueryOperation windowAggregate( + List<Expression> groupingExpressions, + GroupWindow window, + List<Expression> windowProperties, + List<Expression> aggregates, + QueryOperation child) { + + ExpressionResolver resolver = getResolver(child); + ResolvedGroupWindow resolvedWindow = aggregateOperationFactory.createResolvedWindow(window, resolver); + + ExpressionResolver resolverWithWindowReferences = ExpressionResolver.resolverFor( + tableReferenceLookup, + functionCatalog, + child) + .withLocalReferences( + new LocalReferenceExpression( + resolvedWindow.getAlias(), + resolvedWindow.getTimeAttribute().getOutputDataType())) + .build(); + + List<ResolvedExpression> convertedGroupings = resolverWithWindowReferences.resolve(groupingExpressions); + List<ResolvedExpression> convertedAggregates = resolverWithWindowReferences.resolve(aggregates); + List<ResolvedExpression> convertedProperties = resolverWithWindowReferences.resolve(windowProperties); + + return aggregateOperationFactory.createWindowAggregate( + convertedGroupings, + convertedAggregates, + convertedProperties, + resolvedWindow, + child); + } + + public QueryOperation join( + QueryOperation left, + QueryOperation right, + JoinType joinType, + Optional<Expression> condition, + boolean correlated) { + ExpressionResolver resolver = ExpressionResolver.resolverFor(tableReferenceLookup, functionCatalog, left, right) + .build(); + Optional<ResolvedExpression> resolvedCondition = condition.map(expr -> resolveSingleExpression(expr, resolver)); + + return joinOperationFactory.create( + left, + right, + joinType, + resolvedCondition.orElse(valueLiteral(true)), + correlated); + } + + public QueryOperation joinLateral( + QueryOperation left, + Expression tableFunction, + JoinType joinType, + Optional<Expression> condition) { + ExpressionResolver resolver = getResolver(left); + ResolvedExpression resolvedFunction = resolveSingleExpression(tableFunction, resolver); + + QueryOperation temporalTable = calculatedTableFactory.create( + resolvedFunction, + left.getTableSchema().getFieldNames()); + + return join(left, temporalTable, joinType, condition, true); + } + + public Expression resolveExpression(Expression expression, QueryOperation... tableOperation) { + ExpressionResolver resolver = ExpressionResolver.resolverFor( + tableReferenceLookup, + functionCatalog, + tableOperation).build(); + + return resolveSingleExpression(expression, resolver); + } + + private ResolvedExpression resolveSingleExpression(Expression expression, ExpressionResolver resolver) { + List<ResolvedExpression> resolvedExpression = resolver.resolve(Collections.singletonList(expression)); + if (resolvedExpression.size() != 1) { + throw new ValidationException("Expected single expression"); + } else { + return resolvedExpression.get(0); + } + } + + public QueryOperation sort(List<Expression> fields, QueryOperation child) { + + ExpressionResolver resolver = getResolver(child); + List<ResolvedExpression> resolvedFields = resolver.resolve(fields); + + return sortOperationFactory.createSort(resolvedFields, child, resolver.postResolverFactory()); + } + + public QueryOperation limitWithOffset(int offset, QueryOperation child) { + return sortOperationFactory.createLimitWithOffset(offset, child); + } + + public QueryOperation limitWithFetch(int fetch, QueryOperation child) { + return sortOperationFactory.createLimitWithFetch(fetch, child); + } + + public QueryOperation alias(List<Expression> fields, QueryOperation child) { + List<Expression> newFields = AliasOperationUtils.createAliasList(fields, child); + + return project(newFields, child, true); + } + + public QueryOperation filter(Expression condition, QueryOperation child) { + + ExpressionResolver resolver = getResolver(child); + ResolvedExpression resolvedExpression = resolveSingleExpression(condition, resolver); + PlannerExpression convertedCondition = expressionBridge.bridge(resolvedExpression); + if (convertedCondition.resultType() != Types.BOOLEAN()) { + throw new ValidationException("Filter operator requires a boolean expression as input," + + " but $condition is of type " + convertedCondition.resultType()); + } + + return new FilterQueryOperation(resolvedExpression, child); + } + + public QueryOperation distinct(QueryOperation child) { + return new DistinctQueryOperation(child); + } + + public QueryOperation minus(QueryOperation left, QueryOperation right, boolean all) { + return setOperationFactory.create(MINUS, left, right, all); + } + + public QueryOperation intersect(QueryOperation left, QueryOperation right, boolean all) { + return setOperationFactory.create(INTERSECT, left, right, all); + } + + public QueryOperation union(QueryOperation left, QueryOperation right, boolean all) { + return setOperationFactory.create(UNION, left, right, all); + } + + /* Extensions */ + + public QueryOperation map(Expression mapFunction, QueryOperation child) { + + ExpressionResolver resolver = getResolver(child); + Expression resolvedMapFunction = resolveSingleExpression(mapFunction, resolver); + + if (!ApiExpressionUtils.isFunctionOfKind(resolvedMapFunction, FunctionKind.SCALAR)) { + throw new ValidationException("Only ScalarFunction can be used in the map operator."); Review comment: I just noticed that changes made in FLINK-12899 got lost in this PR. Maybe you can go through the class line by line a last time. ---------------------------------------------------------------- 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
