twalthr commented on a change in pull request #11081: [FLINK-16033][table-api] Introduced Java Table API Expression DSL URL: https://github.com/apache/flink/pull/11081#discussion_r384995250
########## File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java ########## @@ -0,0 +1,1286 @@ +/* + * 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.api.internal; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.table.api.Expressions; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.expressions.ApiExpressionUtils; +import org.apache.flink.table.expressions.Expression; +import org.apache.flink.table.expressions.TimeIntervalUnit; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.types.DataType; + +import java.util.Arrays; +import java.util.stream.Stream; + +import static org.apache.flink.table.expressions.ApiExpressionUtils.MILLIS_PER_DAY; +import static org.apache.flink.table.expressions.ApiExpressionUtils.MILLIS_PER_HOUR; +import static org.apache.flink.table.expressions.ApiExpressionUtils.MILLIS_PER_MINUTE; +import static org.apache.flink.table.expressions.ApiExpressionUtils.MILLIS_PER_SECOND; +import static org.apache.flink.table.expressions.ApiExpressionUtils.objectToExpression; +import static org.apache.flink.table.expressions.ApiExpressionUtils.tableRef; +import static org.apache.flink.table.expressions.ApiExpressionUtils.toMilliInterval; +import static org.apache.flink.table.expressions.ApiExpressionUtils.toMonthInterval; +import static org.apache.flink.table.expressions.ApiExpressionUtils.typeLiteral; +import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall; +import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral; +import static org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType; + +//CHECKSTYLE.OFF: AvoidStarImport|ImportOrder +import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.*; +//CHECKSTYLE.ON: AvoidStarImport|ImportOrder + +/** + * These are Java and Scala common operations that can be used to construct an {@link Expression} AST for + * expression operations. + * + * @param <InType> The accepted type of input expressions, it is {@code Expression} for Scala and + * {@code Object} for Java. Generally the expression DSL works on expressions, the + * reason why Java accepts Object is to remove cumbersome call to {@code lit()} for + * literals. Scala alleviates this problem via implicit conversions. + * @param <OutType> The produced type of the DSL. It is {@code ApiExpression} for Java and {@code Expression} + * for Scala. In scala the infix operations are included via implicit conversions. In Java + * we introduced a wrapper that enables the operations without pulling them through the whole stack. + */ +@PublicEvolving +public abstract class BaseExpressions<InType, OutType> { + protected abstract Expression toExpr(); + + protected abstract OutType toApiSpecificExpression(Expression expression); + + /** + * Specifies a name for an expression i.e. a field. + * + * @param name name for one field + * @param extraNames additional names if the expression expands to multiple fields + */ + public OutType as(String name, String... extraNames) { + return toApiSpecificExpression(ApiExpressionUtils.unresolvedCall( + BuiltInFunctionDefinitions.AS, + Stream.concat( + Stream.of(toExpr(), ApiExpressionUtils.valueLiteral(name)), + Stream.of(extraNames).map(ApiExpressionUtils::valueLiteral) + ).toArray(Expression[]::new))); + } + + /** + * Boolean AND in three-valued logic. This is an infix notation. See also + * {@link Expressions#and(Object, Object, Object...)} for prefix notation with multiple arguments. + * + * @see org.apache.flink.table.api.Expressions#and(Object, Object, Object...) + */ + public OutType and(InType other) { + return toApiSpecificExpression(unresolvedCall(AND, toExpr(), objectToExpression(other))); + } + + /** + * Boolean OR in three-valued logic. This is an infix notation. See also + * {@link Expressions#or(Object, Object, Object...)} for prefix notation with multiple arguments. + * + * @see org.apache.flink.table.api.Expressions#or(Object, Object, Object...) + */ + public OutType or(InType other) { + return toApiSpecificExpression(unresolvedCall(OR, toExpr(), objectToExpression(other))); + } + + /** + * Greater than. + */ + public OutType isGreater(InType other) { + return toApiSpecificExpression(unresolvedCall(GREATER_THAN, toExpr(), objectToExpression(other))); + } + + /** + * Greater than or equal. + */ + public OutType isGreaterOrEqual(InType other) { + return toApiSpecificExpression(unresolvedCall(GREATER_THAN_OR_EQUAL, toExpr(), objectToExpression(other))); + } + + /** + * Less than. + */ + public OutType isLess(InType other) { + return toApiSpecificExpression(unresolvedCall(LESS_THAN, toExpr(), objectToExpression(other))); + } + + /** + * Less than or equal. + */ + public OutType isLessOrEqual(InType other) { + return toApiSpecificExpression(unresolvedCall(LESS_THAN_OR_EQUAL, toExpr(), objectToExpression(other))); + } + + /** + * Equals. + */ + public OutType isEqual(InType other) { + return toApiSpecificExpression(unresolvedCall(EQUALS, toExpr(), objectToExpression(other))); + } + + /** + * Not equal. + */ + public OutType isNotEqual(InType other) { + return toApiSpecificExpression(unresolvedCall(NOT_EQUALS, toExpr(), objectToExpression(other))); + } + + /** + * Returns left plus right. + */ + public OutType plus(InType other) { + return toApiSpecificExpression(unresolvedCall(PLUS, toExpr(), objectToExpression(other))); + } + + /** + * Returns left minus right. + */ + public OutType minus(InType other) { + return toApiSpecificExpression(unresolvedCall(MINUS, toExpr(), objectToExpression(other))); + } + + /** + * Returns left divided by right. + */ + public OutType dividedBy(InType other) { + return toApiSpecificExpression(unresolvedCall(DIVIDE, toExpr(), objectToExpression(other))); + } + + /** + * Returns left multiplied by right. + */ + public OutType multipliedBy(InType other) { + return toApiSpecificExpression(unresolvedCall(TIMES, toExpr(), objectToExpression(other))); + } + + /** + * Returns true if the given expression is between lowerBound and upperBound (both inclusive). + * False otherwise. The parameters must be numeric types or identical comparable types. + * + * @param lowerBound numeric or comparable expression + * @param upperBound numeric or comparable expression + */ + public OutType between(InType lowerBound, InType upperBound) { + return toApiSpecificExpression(unresolvedCall( + BETWEEN, + toExpr(), + objectToExpression(lowerBound), + objectToExpression(upperBound))); + } + + /** + * Returns true if the given expression is not between lowerBound and upperBound (both + * inclusive). False otherwise. The parameters must be numeric types or identical + * comparable types. + * + * @param lowerBound numeric or comparable expression + * @param upperBound numeric or comparable expression + */ + public OutType notBetween(InType lowerBound, InType upperBound) { + return toApiSpecificExpression(unresolvedCall( + NOT_BETWEEN, + toExpr(), + objectToExpression(lowerBound), + objectToExpression(upperBound))); + } + + /** + * Ternary conditional operator that decides which of two other expressions should be evaluated + * based on a evaluated boolean condition. + * + * <p>e.g. lit(42).isGreater(5).then("A", "B") leads to "A" + * + * @param ifTrue expression to be evaluated if condition holds + * @param ifFalse expression to be evaluated if condition does not hold + */ + public OutType then(InType ifTrue, InType ifFalse) { + return toApiSpecificExpression(unresolvedCall( + IF, + toExpr(), + objectToExpression(ifTrue), + objectToExpression(ifFalse))); + } + + /** + * Returns true if the given expression is null. + */ + public OutType isNull() { + return toApiSpecificExpression(unresolvedCall(IS_NULL, toExpr())); + } + + /** + * Returns true if the given expression is not null. + */ + public OutType isNotNull() { + return toApiSpecificExpression(unresolvedCall(IS_NOT_NULL, toExpr())); + } + + /** + * Returns true if given boolean expression is true. False otherwise (for null and false). + */ + public OutType isTrue() { + return toApiSpecificExpression(unresolvedCall(IS_TRUE, toExpr())); + } + + /** + * Returns true if given boolean expression is false. False otherwise (for null and true). + */ + public OutType isFalse() { + return toApiSpecificExpression(unresolvedCall(IS_FALSE, toExpr())); + } + + /** + * Returns true if given boolean expression is not true (for null and false). False otherwise. + */ + public OutType isNotTrue() { + return toApiSpecificExpression(unresolvedCall(IS_NOT_TRUE, toExpr())); + } + + /** + * Returns true if given boolean expression is not false (for null and true). False otherwise. + */ + public OutType isNotFalse() { + return toApiSpecificExpression(unresolvedCall(IS_NOT_FALSE, toExpr())); + } + + /** + * Similar to a SQL distinct aggregation clause such as COUNT(DISTINCT a), declares that an + * aggregation function is only applied on distinct input values. + * + * <p>For example: + * <pre> + * {@code + * orders + * .groupBy($("a")) + * .select($("a"), $("b").sum().distinct().as("d")) + * } + * </pre> + */ + public OutType distinct() { + return toApiSpecificExpression(unresolvedCall(DISTINCT, toExpr())); + } + + /** + * Returns the sum of the numeric field across all input values. + * If all values are null, null is returned. + */ + public OutType sum() { + return toApiSpecificExpression(unresolvedCall(SUM, toExpr())); + } + + /** + * Returns the sum of the numeric field across all input values. + * If all values are null, 0 is returned. + */ + public OutType sum0() { + return toApiSpecificExpression(unresolvedCall(SUM0, toExpr())); + } + + /** + * Returns the minimum value of field across all input values. + */ + public OutType min() { + return toApiSpecificExpression(unresolvedCall(MIN, toExpr())); + } + + /** + * Returns the maximum value of field across all input values. + */ + public OutType max() { + return toApiSpecificExpression(unresolvedCall(MAX, toExpr())); + } + + /** + * Returns the number of input rows for which the field is not null. + */ + public OutType count() { + return toApiSpecificExpression(unresolvedCall(COUNT, toExpr())); + } + + /** + * Returns the average (arithmetic mean) of the numeric field across all input values. + */ + public OutType avg() { + return toApiSpecificExpression(unresolvedCall(AVG, toExpr())); + } + + /** + * Returns the population standard deviation of an expression (the square root of varPop()). + */ + public OutType stddevPop() { + return toApiSpecificExpression(unresolvedCall(STDDEV_POP, toExpr())); + } + + /** + * Returns the sample standard deviation of an expression (the square root of varSamp()). + */ + public OutType stddevSamp() { + return toApiSpecificExpression(unresolvedCall(STDDEV_SAMP, toExpr())); + } + + /** + * Returns the population standard variance of an expression. + */ + public OutType varPop() { + return toApiSpecificExpression(unresolvedCall(VAR_POP, toExpr())); + } + + /** + * Returns the sample variance of a given expression. + */ + public OutType varSamp() { + return toApiSpecificExpression(unresolvedCall(VAR_SAMP, toExpr())); + } + + /** + * Returns multiset aggregate of a given expression. + */ + public OutType collect() { + return toApiSpecificExpression(unresolvedCall(COLLECT, toExpr())); + } + + /** + * Converts a value to a given data type. + * + * <p>e.g. "42".cast(DataTypes.INT()) leads to 42. + */ + public OutType cast(DataType toType) { + return toApiSpecificExpression(unresolvedCall(CAST, toExpr(), typeLiteral(toType))); + } + + /** + * @deprecated This method will be removed in future versions as it uses the old type system. It + * is recommended to use {@link #cast(DataType)} instead which uses the new type system + * based on {@link org.apache.flink.table.api.DataTypes}. Please make sure to use either the old + * or the new type system consistently to avoid unintended behavior. See the website documentation + * for more information. + */ + @Deprecated + public OutType cast(TypeInformation<?> toType) { + return toApiSpecificExpression(unresolvedCall(CAST, toExpr(), typeLiteral(fromLegacyInfoToDataType(toType)))); + } + + /** + * Specifies ascending order of an expression i.e. a field for orderBy unresolvedCall. + */ + public OutType asc() { + return toApiSpecificExpression(unresolvedCall(ORDER_ASC, toExpr())); + } + + /** + * Specifies descending order of an expression i.e. a field for orderBy unresolvedCall. + */ + public OutType desc() { + return toApiSpecificExpression(unresolvedCall(ORDER_DESC, toExpr())); + } + + /** + * Returns true if an expression exists in a given list of expressions. This is a shorthand + * for multiple OR conditions. + * + * <p>If the testing set contains null, the result will be null if the element can not be found + * and true if it can be found. If the element is null, the result is always null. + * + * <p>e.g. lit("42").in(1, 2, 3) leads to false. + */ + @SafeVarargs + public final OutType in(InType... elements) { + Expression[] args = Stream.concat( + Stream.of(toExpr()), + Arrays.stream(elements).map(ApiExpressionUtils::objectToExpression)) + .toArray(Expression[]::new); + return toApiSpecificExpression(unresolvedCall(IN, args)); + } + + /** + * Returns true if an expression exists in a given table sub-query. The sub-query table + * must consist of one column. This column must have the same data type as the expression. + * + * <p>Note: This operation is not supported in a streaming environment yet. + */ + public OutType in(Table table) { + return toApiSpecificExpression(unresolvedCall(IN, toExpr(), tableRef(table.toString(), table))); + } + + /** + * Returns the start time (inclusive) of a window when applied on a window reference. + */ + public OutType start() { + return toApiSpecificExpression(unresolvedCall(WINDOW_START, toExpr())); + } + + /** + * Returns the end time (exclusive) of a window when applied on a window reference. + * + * <p>e.g. if a window ends at 10:59:59.999 this property will return 11:00:00.000. + */ + public OutType end() { + return toApiSpecificExpression(unresolvedCall(WINDOW_END, toExpr())); + } + + // scalar functions Review comment: nit: remove it or make it more visible like? ``` // ------ // Scalar function // ---- ``` ---------------------------------------------------------------- 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
