This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 34b6c8fbbc1d6d94d6d76f43bfe52dce2b0d8122 Author: Sergey Nuyanzin <[email protected]> AuthorDate: Sat Apr 25 21:11:43 2026 +0200 [FLINK-35856][table] Upgrade Calcite version to 1.37.0 This closes #28152. Co-authored-by: Michal Stutzmann <[email protected]> --- flink-formats/flink-protobuf/pom.xml | 2 +- flink-table/flink-sql-parser/pom.xml | 7 +- .../src/main/codegen/data/Parser.tdd | 1 + .../src/main/codegen/templates/Parser.jj | 99 ++++++--- .../main/java/org/apache/calcite/sql/SqlJoin.java | 8 +- .../org/apache/calcite/sql/type/SqlTypeFamily.java | 16 +- .../org/apache/calcite/sql/type/SqlTypeName.java | 37 +++- .../flink/sql/parser/FlinkDDLDataTypeTest.java | 4 - .../flink/sql/parser/FlinkSqlParserImplTest.java | 3 + flink-table/flink-table-calcite-bridge/pom.xml | 14 +- .../calcite/rel/metadata/RelMdPredicates.java | 84 +++++--- .../calcite/rel/type/RelDataTypeFactoryImpl.java | 5 +- .../main/java/org/apache/calcite/rex/RexUtil.java | 46 +++- .../org/apache/calcite/runtime/SqlFunctions.java | 164 +++++++++++++- .../main/java/org/apache/calcite/sql/SqlCall.java | 2 +- .../calcite/sql/SqlGroupedWindowFunction.java | 46 ++-- .../apache/calcite/sql/SqlIntervalQualifier.java | 8 +- .../main/java/org/apache/calcite/sql/SqlUtil.java | 2 +- .../apache/calcite/sql/fun/SqlCastFunction.java | 28 ++- .../calcite/sql/fun/SqlMapValueConstructor.java | 45 ++-- .../calcite/sql/type/SqlTypeAssignmentRule.java | 9 + .../calcite/sql/type/SqlTypeFactoryImpl.java | 5 + .../org/apache/calcite/sql/type/SqlTypeUtil.java | 22 +- .../calcite/sql/validate/SqlValidatorImpl.java | 237 +++++++++++++++++---- .../apache/calcite/sql2rel/SqlToRelConverter.java | 106 ++++++--- .../calcite/sql2rel/StandardConvertletTable.java | 16 +- .../nodes/logical/FlinkLogicalOverAggregate.java | 6 + .../src/main/resources/META-INF/NOTICE | 6 +- .../org.codehaus.commons.compiler.ICompilerFactory | 16 ++ .../table/planner/codegen/ExprCodeGenerator.scala | 8 + .../planner/plan/utils/RexDefaultVisitor.scala | 4 + .../planner/plan/utils/RexNodeExtractor.scala | 4 + .../logical/subquery/SubQueryAntiJoinTest.xml | 4 +- .../logical/subquery/SubQuerySemiJoinTest.xml | 4 +- .../logical/subquery/SubQuerySemiJoinTest.scala | 3 - .../src/main/resources/META-INF/NOTICE | 4 +- flink-table/pom.xml | 7 +- 37 files changed, 827 insertions(+), 255 deletions(-) diff --git a/flink-formats/flink-protobuf/pom.xml b/flink-formats/flink-protobuf/pom.xml index c7284fdd126..9ca5eea6762 100644 --- a/flink-formats/flink-protobuf/pom.xml +++ b/flink-formats/flink-protobuf/pom.xml @@ -36,7 +36,7 @@ under the License. <properties> <!-- the same with flink-table/pom.xml --> - <janino.version>3.1.10</janino.version> + <janino.version>3.1.11</janino.version> </properties> <dependencies> diff --git a/flink-table/flink-sql-parser/pom.xml b/flink-table/flink-sql-parser/pom.xml index 56629b88b59..366d6d8ea1f 100644 --- a/flink-table/flink-sql-parser/pom.xml +++ b/flink-table/flink-sql-parser/pom.xml @@ -68,10 +68,11 @@ under the License. <version>${calcite.version}</version> <exclusions> <!-- - "mvn dependency:tree" as of Calcite 1.36.0: + "mvn dependency:tree" as of Calcite 1.37.0: - [INFO] +- org.apache.calcite:calcite-core:jar:1.36.0:compile - [INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.23.0:compile + [INFO] +- org.apache.calcite:calcite-core:jar:1.37.0:compile + [INFO] | +- org.apache.calcite:calcite-linq4j:jar:1.37.0:compile + [INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.25.0:compile [INFO] | +- org.apiguardian:apiguardian-api:jar:1.1.2:compile [INFO] | +- org.checkerframework:checker-qual:jar:3.37.0:compile [INFO] | \- org.apache.commons:commons-math3:jar:3.6.1:runtime diff --git a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd index 74339361d79..1b739369149 100644 --- a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd +++ b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd @@ -185,6 +185,7 @@ "org.apache.calcite.sql.SqlDrop" "org.apache.calcite.sql.SqlExplicitModelOperator" "org.apache.calcite.sql.SqlIntervalLiteral" + "org.apache.calcite.sql.SqlLambda" "org.apache.calcite.runtime.Resources" "java.util.ArrayList" "java.util.Collections" diff --git a/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj b/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj index a9802fd631b..1f0ae3d455c 100644 --- a/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj +++ b/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj @@ -1033,6 +1033,9 @@ void AddArg0(List<SqlNode> list, ExprContext exprContext) : ) ( e = Default() + | + LOOKAHEAD((SimpleIdentifierOrList() | <LPAREN> <RPAREN>) <LAMBDA>) + e = LambdaExpression() | LOOKAHEAD(3) e = TableParam() @@ -1060,6 +1063,9 @@ void AddArg(List<SqlNode> list, ExprContext exprContext) : ) ( e = Default() + | + LOOKAHEAD((SimpleIdentifierOrList() | <LPAREN> <RPAREN>) <LAMBDA>) + e = LambdaExpression() | e = Expression(exprContext) | @@ -3964,6 +3970,43 @@ SqlNode Expression3(ExprContext exprContext) : } } +/** + * Parses a lambda expression. + */ +SqlNode LambdaExpression() : +{ + final SqlNodeList parameters; + final SqlNode expression; + final Span s; +} +{ + parameters = SimpleIdentifierOrListOrEmpty() + <LAMBDA> { s = span(); } + expression = Expression(ExprContext.ACCEPT_NON_QUERY) + { + return new SqlLambda(s.end(this), parameters, expression); + } +} + +/** + * List of simple identifiers in parentheses or empty parentheses or one simple identifier. + * <ul>Examples: + * <li>{@code ()} + * <li>{@code DEPTNO} + * <li>{@code (EMPNO, DEPTNO)} + * </ul> + */ +SqlNodeList SimpleIdentifierOrListOrEmpty() : +{ + SqlNodeList list; +} +{ + LOOKAHEAD(2) + <LPAREN> <RPAREN> { return SqlNodeList.EMPTY; } +| + list = SimpleIdentifierOrList() { return list; } +} + SqlOperator periodOperator() : { } @@ -4756,6 +4799,7 @@ SqlLiteral DateTimeLiteral() : { final String p; final Span s; + boolean local = false; } { <LBRACE_D> <QUOTED_STRING> { @@ -4787,6 +4831,7 @@ SqlLiteral DateTimeLiteral() : return SqlLiteral.createUnknown("DATETIME", p, s.end(this)); } | + LOOKAHEAD(2) <TIME> { s = span(); } p = SimpleStringLiteral() { return SqlLiteral.createUnknown("TIME", p, s.end(this)); } @@ -4796,8 +4841,14 @@ SqlLiteral DateTimeLiteral() : return SqlLiteral.createUnknown("TIMESTAMP", p, s.end(this)); } | - <TIMESTAMP> { s = span(); } <WITH> <LOCAL> <TIME> <ZONE> p = SimpleStringLiteral() { - return SqlLiteral.createUnknown("TIMESTAMP WITH LOCAL TIME ZONE", p, s.end(this)); + LOOKAHEAD(2) + <TIME> { s = span(); } <WITH> ( <LOCAL> { local = true; } )? <TIME> <ZONE> p = SimpleStringLiteral() { + return SqlLiteral.createUnknown("TIME WITH " + (local ? "LOCAL " : "") + "TIME ZONE", p, s.end(this)); + } +| + LOOKAHEAD(2) + <TIMESTAMP> { s = span(); } <WITH> ( <LOCAL> { local = true; } )? <TIME> <ZONE> p = SimpleStringLiteral() { + return SqlLiteral.createUnknown("TIMESTAMP WITH " + (local ? "LOCAL " : "") + "TIME ZONE", p, s.end(this)); } } @@ -4935,7 +4986,6 @@ SqlCall ArrayLiteral() : SqlNode MapConstructor() : { SqlNodeList args; - SqlNode e; final Span s; } { @@ -6084,7 +6134,6 @@ SqlTypeNameSpec DateTimeTypeName() : { int precision = -1; SqlTypeName typeName; - boolean withLocalTimeZone = false; final Span s; } { @@ -6096,25 +6145,15 @@ SqlTypeNameSpec DateTimeTypeName() : LOOKAHEAD(2) <TIME> { s = span(); } precision = PrecisionOpt() - withLocalTimeZone = TimeZoneOpt() + typeName = TimeZoneOpt(true) { - if (withLocalTimeZone) { - typeName = SqlTypeName.TIME_WITH_LOCAL_TIME_ZONE; - } else { - typeName = SqlTypeName.TIME; - } return new SqlBasicTypeNameSpec(typeName, precision, s.end(this)); } | <TIMESTAMP> { s = span(); } precision = PrecisionOpt() - withLocalTimeZone = TimeZoneOpt() + typeName = TimeZoneOpt(false) { - if (withLocalTimeZone) { - typeName = SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE; - } else { - typeName = SqlTypeName.TIMESTAMP; - } return new SqlBasicTypeNameSpec(typeName, precision, s.end(this)); } } @@ -6135,23 +6174,22 @@ int PrecisionOpt() : /** * Parse a time zone suffix for DateTime types. According to SQL-2011, -* "with time zone" and "without time zone" belong to standard SQL but we -* only implement the "without time zone". -* -* <p>We also support "with local time zone". -* -* @return true if this is "with local time zone". +* "with time zone" and "without time zone" belong to standard SQL. +* We also support "with local time zone". */ -boolean TimeZoneOpt() : +SqlTypeName TimeZoneOpt(boolean timeType) : { + boolean local = false; } { LOOKAHEAD(3) - <WITHOUT> <TIME> <ZONE> { return false; } + <WITHOUT> <TIME> <ZONE> { return timeType ? SqlTypeName.TIME : SqlTypeName.TIMESTAMP; } | - <WITH> <LOCAL> <TIME> <ZONE> { return true; } + <WITH> ( <LOCAL> { local = true; } )? <TIME> <ZONE> { + return timeType ? (local ? SqlTypeName.TIME_WITH_LOCAL_TIME_ZONE : SqlTypeName.TIME_TZ) + : (local ? SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE : SqlTypeName.TIMESTAMP_TZ); } | - { return false; } + { return timeType ? SqlTypeName.TIME : SqlTypeName.TIMESTAMP; } } /** @@ -6192,6 +6230,7 @@ SqlNode BuiltinFunctionCall() : final SqlNode node; final SqlLiteral style; // mssql convert 'style' operand final SqlFunction f; + final SqlNode format; } { //~ FUNCTIONS WITH SPECIAL SYNTAX --------------------------------------- @@ -6208,6 +6247,7 @@ SqlNode BuiltinFunctionCall() : | <INTERVAL> e = IntervalQualifier() { args.add(e); } ) + [ <FORMAT> format = StringLiteral() { args.add(format); } ] <RPAREN> { return f.createCall(s.end(this), args); } @@ -6747,6 +6787,12 @@ List<SqlNode> JsonNameAndValue() : } ( <VALUE> + | + <COMMA> { + if (kvMode) { + throw SqlUtil.newContextException(getPos(), RESOURCE.illegalComma()); + } + } | <COLON> { if (kvMode) { @@ -8848,6 +8894,7 @@ void NonReservedKeyWord2of3() : | < NE2: "!=" > | < PLUS: "+" > | < MINUS: "-" > +| < LAMBDA: "->" > | < STAR: "*" > | < SLASH: "/" > | < PERCENT_REMAINDER: "%" > diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/SqlJoin.java b/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/SqlJoin.java index ca7766b33fb..455255cd6cc 100644 --- a/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/SqlJoin.java +++ b/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/SqlJoin.java @@ -17,7 +17,6 @@ package org.apache.calcite.sql; -import com.google.common.base.Preconditions; import org.apache.calcite.sql.parser.SqlParserPos; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.calcite.sql.util.SqlString; @@ -28,6 +27,7 @@ import org.checkerframework.checker.nullness.qual.Nullable; import java.util.List; import java.util.function.UnaryOperator; +import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; /** Parse tree node representing a {@code JOIN} clause. */ @@ -70,7 +70,7 @@ public class SqlJoin extends SqlCall { this.conditionType = requireNonNull(conditionType, "conditionType"); this.condition = condition; - Preconditions.checkArgument(natural.getTypeName() == SqlTypeName.BOOLEAN); + checkArgument(natural.getTypeName() == SqlTypeName.BOOLEAN); conditionType.getValueAs(JoinConditionType.class); joinType.getValueAs(JoinType.class); } @@ -235,6 +235,9 @@ public class SqlJoin extends SqlCall { case LEFT_SEMI_JOIN: writer.sep(join.isNatural() ? "NATURAL LEFT SEMI JOIN" : "LEFT SEMI JOIN"); break; + case LEFT_ANTI_JOIN: + writer.sep(join.isNatural() ? "NATURAL LEFT ANTI JOIN" : "LEFT ANTI JOIN"); + break; case RIGHT: writer.sep(join.isNatural() ? "NATURAL RIGHT JOIN" : "RIGHT JOIN"); break; @@ -283,6 +286,7 @@ public class SqlJoin extends SqlCall { null, null, null, + null, SqlNodeList.EMPTY); return selectWrapper.toSqlString(transform); } diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeFamily.java b/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeFamily.java index 94cca219d92..b755fc4d6f4 100644 --- a/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeFamily.java +++ b/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeFamily.java @@ -78,6 +78,7 @@ public enum SqlTypeFamily implements RelDataTypeFamily { COLUMN_LIST, GEO, VARIANT, + FUNCTION, /** Like ANY, but do not even validate the operand. It may not be an expression. */ IGNORE; @@ -173,10 +174,15 @@ public enum SqlTypeFamily implements RelDataTypeFamily { case DATE: return ImmutableList.of(SqlTypeName.DATE); case TIME: - return ImmutableList.of(SqlTypeName.TIME, SqlTypeName.TIME_WITH_LOCAL_TIME_ZONE); + return ImmutableList.of( + SqlTypeName.TIME, + SqlTypeName.TIME_WITH_LOCAL_TIME_ZONE, + SqlTypeName.TIME_TZ); case TIMESTAMP: return ImmutableList.of( - SqlTypeName.TIMESTAMP, SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE); + SqlTypeName.TIMESTAMP, + SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE, + SqlTypeName.TIMESTAMP_TZ); case BOOLEAN: return SqlTypeName.BOOLEAN_TYPES; case INTERVAL_YEAR_MONTH: @@ -211,6 +217,8 @@ public enum SqlTypeFamily implements RelDataTypeFamily { return ImmutableList.of(SqlTypeName.CURSOR); case COLUMN_LIST: return ImmutableList.of(SqlTypeName.COLUMN_LIST); + case FUNCTION: + return ImmutableList.of(SqlTypeName.FUNCTION); case VARIANT: return ImmutableList.of(SqlTypeName.VARIANT); default: @@ -269,6 +277,10 @@ public enum SqlTypeFamily implements RelDataTypeFamily { return factory.createSqlType(SqlTypeName.CURSOR); case COLUMN_LIST: return factory.createSqlType(SqlTypeName.COLUMN_LIST); + case FUNCTION: + return factory.createFunctionSqlType( + factory.createStructType(ImmutableList.of(), ImmutableList.of()), + factory.createSqlType(SqlTypeName.ANY)); default: return null; } diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeName.java b/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeName.java index 7f0e874d805..5f76b4493b1 100644 --- a/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeName.java +++ b/flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/type/SqlTypeName.java @@ -69,10 +69,13 @@ public enum SqlTypeName { DATE(PrecScale.NO_NO, false, Types.DATE, SqlTypeFamily.DATE), TIME(PrecScale.NO_NO | PrecScale.YES_NO, false, Types.TIME, SqlTypeFamily.TIME), TIME_WITH_LOCAL_TIME_ZONE( - PrecScale.NO_NO | PrecScale.YES_NO, false, Types.OTHER, SqlTypeFamily.TIME), + PrecScale.NO_NO | PrecScale.YES_NO, false, Types.TIME, SqlTypeFamily.TIME), + TIME_TZ(PrecScale.NO_NO | PrecScale.YES_NO, false, Types.TIME, SqlTypeFamily.TIME), TIMESTAMP(PrecScale.NO_NO | PrecScale.YES_NO, false, Types.TIMESTAMP, SqlTypeFamily.TIMESTAMP), TIMESTAMP_WITH_LOCAL_TIME_ZONE( PrecScale.NO_NO | PrecScale.YES_NO, false, Types.TIMESTAMP, SqlTypeFamily.TIMESTAMP), + TIMESTAMP_TZ( + PrecScale.NO_NO | PrecScale.YES_NO, false, Types.TIMESTAMP, SqlTypeFamily.TIMESTAMP), INTERVAL_YEAR(PrecScale.NO_NO, false, Types.OTHER, SqlTypeFamily.INTERVAL_YEAR_MONTH), INTERVAL_YEAR_MONTH(PrecScale.NO_NO, false, Types.OTHER, SqlTypeFamily.INTERVAL_YEAR_MONTH), INTERVAL_MONTH(PrecScale.NO_NO, false, Types.OTHER, SqlTypeFamily.INTERVAL_YEAR_MONTH), @@ -158,6 +161,7 @@ public enum SqlTypeName { */ GEOMETRY(PrecScale.NO_NO, false, ExtraSqlTypes.GEOMETRY, SqlTypeFamily.GEO), MEASURE(PrecScale.NO_NO, true, Types.OTHER, SqlTypeFamily.ANY), + FUNCTION(PrecScale.NO_NO, true, Types.OTHER, SqlTypeFamily.FUNCTION), SARG(PrecScale.NO_NO, true, Types.OTHER, SqlTypeFamily.ANY), /** * VARIANT data type, a dynamically-typed value that can have at runtime any of the other data @@ -173,7 +177,7 @@ public enum SqlTypeName { public static final int DEFAULT_INTERVAL_START_PRECISION = 2; public static final int DEFAULT_INTERVAL_FRACTIONAL_SECOND_PRECISION = 6; public static final int MIN_INTERVAL_START_PRECISION = 1; - public static final int MIN_INTERVAL_FRACTIONAL_SECOND_PRECISION = 1; + public static final int MIN_INTERVAL_FRACTIONAL_SECOND_PRECISION = 0; public static final int MAX_INTERVAL_START_PRECISION = 10; public static final int MAX_INTERVAL_FRACTIONAL_SECOND_PRECISION = 9; @@ -220,7 +224,9 @@ public enum SqlTypeName { INTERVAL_MINUTE_SECOND, INTERVAL_SECOND, TIME_WITH_LOCAL_TIME_ZONE, + TIME_TZ, TIMESTAMP_WITH_LOCAL_TIME_ZONE, + TIMESTAMP_TZ, FLOAT, MULTISET, DISTINCT, @@ -258,8 +264,18 @@ public enum SqlTypeName { DATE, TIME, TIME_WITH_LOCAL_TIME_ZONE, + TIME_TZ, TIMESTAMP, - TIMESTAMP_WITH_LOCAL_TIME_ZONE); + TIMESTAMP_WITH_LOCAL_TIME_ZONE, + TIMESTAMP_TZ); + + /** Types that contain time zone information. */ + public static final List<SqlTypeName> TZ_TYPES = + ImmutableList.of( + TIME_WITH_LOCAL_TIME_ZONE, + TIME_TZ, + TIMESTAMP_WITH_LOCAL_TIME_ZONE, + TIMESTAMP_TZ); public static final Set<SqlTypeName> YEAR_INTERVAL_TYPES = Sets.immutableEnumSet( @@ -369,7 +385,12 @@ public enum SqlTypeName { * or throws {@link IllegalArgumentException}; never returns null. */ public static SqlTypeName lookup(String tag) { - String tag2 = tag.replace(' ', '_'); + // Special handling for TIME WITH TIME ZONE and + // TIMESTAMP WITH TIME ZONE, whose type names are TIME_TZ and TIMESTAMP_TZ. + // We know that the type name is always uppercase, because it is + // inserted in the tag by the parser. + final String tag1 = tag.replace("WITH TIME ZONE", "TZ"); + final String tag2 = tag1.replace(' ', '_'); return valueOf(tag2); } @@ -398,8 +419,8 @@ public enum SqlTypeName { * true</code>, because the VARCHAR type allows a precision parameter, as in <code>VARCHAR(10) * </code>. * <li><code>Varchar.allowsPrecScale(true, true)</code> returns <code> - * true</code>, because the VARCHAR type does not allow a precision and a scale parameter, as in - * <code>VARCHAR(10, 4)</code>. + * false</code>, because the VARCHAR type does not allow a precision and a scale parameter, as + * in <code>VARCHAR(10, 4)</code>. * <li><code>allowsPrecScale(false, true)</code> returns <code>false</code> for every type. * </ul> * @@ -419,7 +440,7 @@ public enum SqlTypeName { return special; } - /** Returns the ordinal from {@link Types} corresponding to this SqlTypeName. */ + /** Returns the ordinal from {@link java.sql.Types} corresponding to this SqlTypeName. */ public int getJdbcOrdinal() { return jdbcOrdinal; } @@ -826,8 +847,10 @@ public enum SqlTypeName { case BINARY: case TIME: case TIME_WITH_LOCAL_TIME_ZONE: + case TIME_TZ: case TIMESTAMP: case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + case TIMESTAMP_TZ: return 1; case INTERVAL_YEAR: case INTERVAL_YEAR_MONTH: diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkDDLDataTypeTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkDDLDataTypeTest.java index 650c8d70433..d6ff47750bc 100644 --- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkDDLDataTypeTest.java +++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkDDLDataTypeTest.java @@ -328,10 +328,6 @@ class FlinkDDLDataTypeTest { + "', `age` INTEGER 'This is comment', `updated` BOOLEAN NOT NULL 'This as well' >"), // Test parse throws error. - createArgumentsTestItem( - "TIMESTAMP WITH ^TIME^ ZONE", "(?s).*Encountered \"TIME\" at .*"), - createArgumentsTestItem( - "TIMESTAMP(3) WITH ^TIME^ ZONE", "(?s).*Encountered \"TIME\" at .*"), createArgumentsTestItem( "^NULL^", "(?s).*Incorrect syntax near the keyword 'NULL' at line 2, column 6..*"), diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java index c147caec32f..1932f2f4a90 100644 --- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java +++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java @@ -70,6 +70,9 @@ class FlinkSqlParserImplTest extends SqlParserTest { @Test void testMapQueryConstructor() {} + @Test + void testMultisetQueryConstructor() {} + @Test void testShowCatalogs() { sql("show catalogs").ok("SHOW CATALOGS"); diff --git a/flink-table/flink-table-calcite-bridge/pom.xml b/flink-table/flink-table-calcite-bridge/pom.xml index be8f6441b27..2de3dced880 100644 --- a/flink-table/flink-table-calcite-bridge/pom.xml +++ b/flink-table/flink-table-calcite-bridge/pom.xml @@ -45,15 +45,15 @@ under the License. <version>${calcite.version}</version> <exclusions> <!-- - "mvn dependency:tree" as of Calcite 1.36.0: - [INFO] +- org.apache.calcite:calcite-core:jar:1.36.0:compile - [INFO] | +- org.apache.calcite:calcite-linq4j:jar:1.36.0:compile + "mvn dependency:tree" as of Calcite 1.37.0: + [INFO] +- org.apache.calcite:calcite-core:jar:1.37.0:compile + [INFO] | +- org.apache.calcite:calcite-linq4j:jar:1.37.0:compile [INFO] | +- org.locationtech.jts:jts-core:jar:1.19.0:compile - [INFO] | +- com.fasterxml.jackson.core:jackson-annotations:jar:2.20.1:compile - [INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.23.0:compile + [INFO] | +- com.fasterxml.jackson.core:jackson-annotations:jar:2.21:compile + [INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.25.0:compile [INFO] | +- org.apiguardian:apiguardian-api:jar:1.1.2:compile - [INFO] | +- com.fasterxml.jackson.core:jackson-core:jar:2.20.1:compile - [INFO] | +- com.fasterxml.jackson.core:jackson-databind:jar:2.20.1:compile + [INFO] | +- com.fasterxml.jackson.core:jackson-core:jar:2.21.3:compile + [INFO] | +- com.fasterxml.jackson.core:jackson-databind:jar:2.21.3:compile [INFO] | +- commons-codec:commons-codec:jar:1.15:runtime [INFO] | +- org.apache.commons:commons-math3:jar:3.6.1:runtime [INFO] | \- commons-io:commons-io:jar:2.15.1:runtime diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java index dffe86ad1d4..de3d4a6deaa 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java @@ -27,6 +27,7 @@ import org.apache.calcite.plan.Strong; import org.apache.calcite.plan.volcano.RelSubset; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.core.Aggregate; +import org.apache.calcite.rel.core.AggregateCall; import org.apache.calcite.rel.core.Correlate; import org.apache.calcite.rel.core.Exchange; import org.apache.calcite.rel.core.Filter; @@ -40,6 +41,7 @@ import org.apache.calcite.rel.core.Sort; import org.apache.calcite.rel.core.TableModify; import org.apache.calcite.rel.core.TableScan; import org.apache.calcite.rel.core.Union; +import org.apache.calcite.rel.core.Values; import org.apache.calcite.rex.RexBuilder; import org.apache.calcite.rex.RexCall; import org.apache.calcite.rex.RexExecutor; @@ -52,7 +54,7 @@ import org.apache.calcite.rex.RexUnknownAs; import org.apache.calcite.rex.RexUtil; import org.apache.calcite.rex.RexVisitorImpl; import org.apache.calcite.sql.SqlKind; -import org.apache.calcite.sql.SqlOperator; +import org.apache.calcite.sql.fun.SqlInternalOperators; import org.apache.calcite.sql.fun.SqlStdOperatorTable; import org.apache.calcite.util.BitSets; import org.apache.calcite.util.Bug; @@ -83,15 +85,6 @@ import static java.util.Objects.requireNonNull; /** * Utility to infer Predicates that are applicable above a RelNode. * - * <p>The class was copied over because of * CALCITE-6317 * and should be removed after upgraded to - * calcite-1.37.0. - * - * <p>FLINK modifications are at lines - * - * <ol> - * <li>Port fix of CALCITE-6317 (Calcite 1.37.0): Lines 333~357, 402~404 - * </ol> - * * <p>This is currently used by {@link * org.apache.calcite.rel.rules.JoinPushTransitivePredicatesRule} to infer <em>Predicates</em> that * can be inferred from one side of a Join to the other. @@ -126,6 +119,12 @@ import static java.util.Objects.requireNonNull; * <li>For Full Outer Joins no predicates are pulledUp or inferred. * </ul> * </ol> + * + * The class contains CALCITE-6599 fix from 1.39.0, should be dropped after upgrade to that version + * + * <p>Lines 181 ~ 217 + * + * <p>Lines 560 ~ 562 disable CALCITE-6044 optimization */ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicates> { public static final RelMetadataProvider SOURCE = @@ -179,6 +178,7 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat * </ol> */ public RelOptPredicateList getPredicates(Project project, RelMetadataQuery mq) { + // FLINK MODIFICATION BEGIN final RelNode input = project.getInput(); final RexBuilder rexBuilder = project.getCluster().getRexBuilder(); final RelOptPredicateList inputInfo = mq.getPulledUpPredicates(input); @@ -196,20 +196,10 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat int sIdx = ((RexInputRef) expr.e).getIndex(); m.set(sIdx, expr.i); columnsMappedBuilder.set(sIdx); - // Project can also generate constants. We need to include them. - } else if (RexLiteral.isNullLiteral(expr.e)) { - projectPullUpPredicates.add( - rexBuilder.makeCall( - SqlStdOperatorTable.IS_NULL, - rexBuilder.makeInputRef(project, expr.i))); } else if (RexUtil.isConstant(expr.e)) { - final List<RexNode> args = - ImmutableList.of(rexBuilder.makeInputRef(project, expr.i), expr.e); - final SqlOperator op = - args.get(0).getType().isNullable() || args.get(1).getType().isNullable() - ? SqlStdOperatorTable.IS_NOT_DISTINCT_FROM - : SqlStdOperatorTable.EQUALS; - projectPullUpPredicates.add(rexBuilder.makeCall(op, args)); + // Project can also generate constants (including NULL). We need to + // include them. + projectPullUpPredicates.add(eqConstant(project, rexBuilder, expr.i, expr.e)); } } @@ -224,6 +214,23 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat } } return RelOptPredicateList.of(rexBuilder, projectPullUpPredicates); + // FLINK MODIFICATION END + } + + /** + * Returns a predicate that field {@code i} of relational expression {@code r} is equal to a + * constant expression (using {@code IS NOT DISTINCT FROM} if the expression is nullable, or + * {@code IS NULL} if it is literal null. + */ + private static RexNode eqConstant(RelNode r, RexBuilder rexBuilder, int i, RexNode e) { + final RexInputRef ref = rexBuilder.makeInputRef(r, i); + if (RexLiteral.isNullLiteral(e)) { + return rexBuilder.makeCall(SqlStdOperatorTable.IS_NULL, ref); + } else if (ref.getType().isNullable() || e.getType().isNullable()) { + return rexBuilder.makeCall(SqlStdOperatorTable.IS_NOT_DISTINCT_FROM, ref, e); + } else { + return rexBuilder.makeCall(SqlStdOperatorTable.EQUALS, ref, e); + } } /** @@ -331,7 +338,6 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat return joinInference.inferPredicates(false); } - // FLINK MODIFICATION BEGIN /** * Check whether the fields specified by the predicateColumns appear in all the groupSets of the * aggregate. @@ -355,8 +361,6 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat return true; } - // FLINK MODIFICATION END - /** * Infers predicates for an Aggregate. * @@ -384,7 +388,7 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat // it is not valid to pull up predicates. In particular, consider the // predicate "false": it is valid on all input rows (trivially - there are // no rows!) but not on the output (there is one row). - return RelOptPredicateList.EMPTY; + return RelOptPredicateList.of(rexBuilder, aggPullUpPredicates); } Mapping m = Mappings.create( @@ -400,13 +404,23 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat for (RexNode r : inputInfo.pulledUpPredicates) { ImmutableBitSet rCols = RelOptUtil.InputFinder.bits(r); - // FLINK MODIFICATION BEGIN if (groupKeys.contains(rCols) && this.allGroupSetsOverlap(rCols, agg)) { - // FLINK MODIFICATION END r = r.accept(new RexPermuteInputsShuttle(m, input)); aggPullUpPredicates.add(r); } } + + i = agg.getGroupCount(); + for (AggregateCall aggregateCall : agg.getAggCallList()) { + if (aggregateCall.getAggregation() == SqlInternalOperators.LITERAL_AGG) { + // The query + // SELECT x, LITERAL_AGG[42]() AS y FROM t GROUP BY x + // has predicate "y = 42" + aggPullUpPredicates.add( + eqConstant(agg, rexBuilder, i, aggregateCall.rexList.get(0))); + } + ++i; + } return RelOptPredicateList.of(rexBuilder, aggPullUpPredicates); } @@ -536,6 +550,18 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat return mq.getPulledUpPredicates(input); } + /** + * Infers predicates for a Values. + * + * <p>The predicates on {@code T (x, y, z)} with rows {@code (1, 2, null), (1, 2, null), (5, 2, + * null)} are {@code 'y = 2'} and {@code 'z is null'}. + */ + public RelOptPredicateList getPredicates(Values values, RelMetadataQuery mq) { + // FLINK MODIFICATION BEGIN + return RelOptPredicateList.EMPTY; + // FLINK MODIFICATION END + } + // CHECKSTYLE: IGNORE 1 /** * Returns the {@link BuiltInMetadata.Predicates#getPredicates()} statistic. diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/type/RelDataTypeFactoryImpl.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/type/RelDataTypeFactoryImpl.java index 33c664e7d70..6dc339db70f 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/type/RelDataTypeFactoryImpl.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/type/RelDataTypeFactoryImpl.java @@ -58,7 +58,8 @@ import static org.apache.calcite.util.ReflectUtil.isStatic; * <p>FLINK modifications are at lines * * <ol> - * <li>Should be removed after fixing CALCITE-5199: Lines 242-244 + * <li>Should be removed after fixing CALCITE-5199: Lines 243-245 + * <li>Might be a subject to reconsider after bump to Calcite 1.38.0: Lines 568-570 * </ol> */ public abstract class RelDataTypeFactoryImpl implements RelDataTypeFactory { @@ -564,7 +565,9 @@ public abstract class RelDataTypeFactoryImpl implements RelDataTypeFactory { case REAL: return createSqlType(SqlTypeName.DECIMAL, 14, 7); case FLOAT: + // FLINK MODIFICATION BEGIN return createSqlType(SqlTypeName.DECIMAL, 14, 7); + // FLINK MODIFICATION END case DOUBLE: // the default max precision is 19, so this is actually DECIMAL(19, 15) // but derived system can override the max precision/scale. diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java index 047b5df5494..77cb8af4323 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java @@ -72,13 +72,14 @@ import java.util.function.Predicate; import static com.google.common.collect.ImmutableList.toImmutableList; import static java.util.Objects.requireNonNull; +import static org.apache.calcite.rel.type.RelDataType.PRECISION_NOT_SPECIFIED; /** * Default implementation of {@link org.apache.calcite.rex.RexUtil}, the class was copied over * because of current Calcite way of inferring constants from IS NOT DISTINCT FROM clashes with * filter push down. * - * <p>Lines 399 ~ 401, Use Calcite 1.32.0 behavior for {@link RexUtil#gatherConstraints(Class, + * <p>Lines 402 ~ 404, Use Calcite 1.32.0 behavior for {@link RexUtil#gatherConstraints(Class, * RexNode, Map, Set, RexBuilder)}. */ public class RexUtil { @@ -797,6 +798,16 @@ public class RexUtil { // "<expr>.FIELD" is constant iff "<expr>" is constant. return fieldAccess.getReferenceExpr().accept(this); } + + @Override + public Boolean visitLambda(RexLambda lambda) { + return false; + } + + @Override + public Boolean visitLambdaRef(RexLambdaRef lambdaRef) { + return false; + } } /** @@ -1647,7 +1658,8 @@ public class RexUtil { * * @param source source type * @param target target type - * @return true iff the conversion is a loss-less cast + * @return 'true' when the conversion can certainly be determined to be loss-less cast, but may + * return 'false' for some lossless casts. */ @API(since = "1.22", status = API.Status.EXPERIMENTAL) public static boolean isLosslessCast(RelDataType source, RelDataType target) { @@ -1671,7 +1683,8 @@ public class RexUtil { if (source.getScale() != -1 && source.getScale() != 0) { sourceLength += source.getScale() + 1; // include decimal mark } - return target.getPrecision() >= sourceLength; + final int targetPrecision = target.getPrecision(); + return targetPrecision == PRECISION_NOT_SPECIFIED || targetPrecision >= sourceLength; } // Return FALSE by default return false; @@ -2861,6 +2874,33 @@ public class RexUtil { } } + /** + * Visitor that collects all the top level SubQueries {@link RexSubQuery} in a projection list + * of a given {@link Project}. + */ + public static class SubQueryCollector extends RexVisitorImpl<Void> { + private List<RexSubQuery> subQueries; + + private SubQueryCollector() { + super(true); + this.subQueries = new ArrayList<>(); + } + + @Override + public Void visitSubQuery(RexSubQuery subQuery) { + subQueries.add(subQuery); + return null; + } + + public static List<RexSubQuery> collect(Project project) { + SubQueryCollector subQueryCollector = new SubQueryCollector(); + for (RexNode node : project.getProjects()) { + node.accept(subQueryCollector); + } + return subQueryCollector.subQueries; + } + } + /** * Visitor that throws {@link org.apache.calcite.util.Util.FoundOne} if applied to an expression * that contains a {@link RexSubQuery}. diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/runtime/SqlFunctions.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/runtime/SqlFunctions.java index 8979735e092..909e0028ae9 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/runtime/SqlFunctions.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/runtime/SqlFunctions.java @@ -39,6 +39,7 @@ import org.apache.calcite.linq4j.function.Deterministic; import org.apache.calcite.linq4j.function.Experimental; import org.apache.calcite.linq4j.function.Function1; import org.apache.calcite.linq4j.function.NonDeterministic; +import org.apache.calcite.linq4j.function.Predicate1; import org.apache.calcite.linq4j.tree.Primitive; import org.apache.calcite.rel.type.TimeFrame; import org.apache.calcite.rel.type.TimeFrameSet; @@ -399,10 +400,18 @@ public class SqlFunctions { * else returns a Pattern object. */ private Pattern validateRegexPattern(String regex, String methodName) { + return validateRegexPattern(regex, methodName, 0); + } + + /** + * Validate regex arguments in REGEXP_* fns, throws an exception for invalid regex patterns, + * else returns a Pattern object. + */ + private Pattern validateRegexPattern(String regex, String methodName, int flags) { try { // Uses java.util.regex as a standard for regex processing // in Calcite instead of RE2 used by BigQuery/GoogleSQL - return cache.getUnchecked(new Key(0, regex)); + return cache.getUnchecked(new Key(flags, regex)); } catch (UncheckedExecutionException e) { if (e.getCause() instanceof PatternSyntaxException) { throw RESOURCE.invalidRegexInputForRegexpFunctions( @@ -499,6 +508,17 @@ public class SqlFunctions { return pattern.matcher(value).find(); } + /** + * SQL {@code REGEXP_LIKE(value, regexp, flags)} function. Throws a runtime exception for + * invalid regular expressions. + */ + @SuppressWarnings("unused") + public boolean regexpLike(String value, String regex, String stringFlags) { + final Pattern pattern = + validateRegexPattern(regex, "REGEXP_LIKE", makeRegexpFlags(stringFlags)); + return pattern.matcher(value).find(); + } + /** * SQL {@code REGEXP_EXTRACT(value, regexp)} function. Returns NULL if there is no match. * Returns an exception if regex is invalid. Uses position=1 and occurrence=1 as default @@ -711,8 +731,16 @@ public class SqlFunctions { flags |= Pattern.DOTALL; break; case 'm': + // PostgreSQL should actually interpret m to be a synonym for n, but this is + // relaxed for consistency. flags |= Pattern.MULTILINE; break; + case 's': + // This flag is in PostgreSQL but doesn't apply to other libraries. This is + // relaxed + // for consistency. + flags &= ~Pattern.DOTALL; + break; default: throw RESOURCE.invalidInputForRegexpReplace(stringFlags).ex(); } @@ -1004,16 +1032,36 @@ public class SqlFunctions { /** SQL SUBSTRING(string FROM ...) function. */ public static String substring(String c, int s) { - final int s0 = s - 1; - if (s0 <= 0) { + if (s <= 1) { return c; } if (s > c.length()) { return ""; } + final int s0 = s - 1; return c.substring(s0); } + // Clamp very large long values to integer values. + // Used by the substring functions. + // Java strings do not support long indexes anyway, + // so this is most likely a safe approximation. + // But if a string has more than 2^31 characters + // the result of calling String.substring will be wrong anyway. + static int clamp(long s) { + if (s < Integer.MIN_VALUE) { + return Integer.MIN_VALUE; + } + if (s > Integer.MAX_VALUE) { + return Integer.MAX_VALUE; + } + return (int) s; + } + + public static String substring(String c, long s) { + return substring(c, clamp(s)); + } + /** SQL SUBSTRING(string FROM ... FOR ...) function. */ public static String substring(String c, int s, int l) { int lc = c.length(); @@ -1031,15 +1079,27 @@ public class SqlFunctions { return c.substring(s0, (int) e0); } + public static String substring(String c, int s, long l) { + return substring(c, s, clamp(l)); + } + + public static String substring(String c, long s, int l) { + return substring(c, clamp(s), l); + } + + public static String substring(String c, long s, long l) { + return substring(c, clamp(s), clamp(l)); + } + /** SQL SUBSTRING(binary FROM ...) function for binary. */ public static ByteString substring(ByteString c, int s) { - final int s0 = s - 1; - if (s0 <= 0) { + if (s <= 1) { return c; } if (s > c.length()) { return ByteString.EMPTY; } + final int s0 = s - 1; return c.substring(s0); } @@ -2732,7 +2792,7 @@ public class SqlFunctions { return Math.pow(b0.doubleValue(), b1.doubleValue()); } - // LN, LOG, LOG10 + // LN, LOG, LOG10, LOG2 /** SQL {@code LOG(number, number2)} function applied to double values. */ public static double log(double d0, double d1) { @@ -2754,6 +2814,16 @@ public class SqlFunctions { return Math.log(d0.doubleValue()) / Math.log(d1.doubleValue()); } + /** SQL {@code LOG2(number)} function applied to double values. */ + public static @Nullable Double log2(double number) { + return (number <= 0) ? null : log(number, 2); + } + + /** SQL {@code LOG2(number)} function applied to BigDecimal values. */ + public static @Nullable Double log2(BigDecimal number) { + return log2(number.doubleValue()); + } + // MOD /** SQL <code>MOD</code> operator applied to byte values. */ @@ -3960,7 +4030,7 @@ public class SqlFunctions { return sb.toString(); } - public String formatTimestamp(DataContext ctx, String fmtString, long timestamp) { + public String formatTimestamp(String fmtString, long timestamp) { return internalFormatDatetime(fmtString, internalToTimestamp(timestamp)); } @@ -3974,11 +4044,47 @@ public class SqlFunctions { return sb.toString().trim(); } - public String formatDate(DataContext ctx, String fmtString, int date) { + public int toDate(String dateString, String fmtString) { + return toInt(new java.sql.Date(internalToDateTime(dateString, fmtString))); + } + + public long toTimestamp(String timestampString, String fmtString) { + return toLong(new java.sql.Timestamp(internalToDateTime(timestampString, fmtString))); + } + + private long internalToDateTime(String dateString, String fmtString) { + final ParsePosition pos = new ParsePosition(0); + + sb.setLength(0); + withElements( + FormatModels.POSTGRESQL, + fmtString, + elements -> elements.forEach(element -> element.toPattern(sb))); + final String dateFormatString = sb.toString().trim(); + + final SimpleDateFormat sdf = new SimpleDateFormat(dateFormatString, Locale.ENGLISH); + final Date date = sdf.parse(dateString, pos); + if (pos.getErrorIndex() >= 0 || pos.getIndex() != dateString.length()) { + SQLException e = + new SQLException( + String.format( + Locale.ROOT, + "Invalid format: '%s' for datetime string: '%s'.", + fmtString, + dateString)); + throw Util.toUnchecked(e); + } + + @SuppressWarnings("JavaUtilDate") + final long millisSinceEpoch = date.getTime(); + return millisSinceEpoch; + } + + public String formatDate(String fmtString, int date) { return internalFormatDatetime(fmtString, internalToDate(date)); } - public String formatTime(DataContext ctx, String fmtString, int time) { + public String formatTime(String fmtString, int time) { return internalFormatDatetime(fmtString, internalToTime(time)); } } @@ -4457,6 +4563,14 @@ public class SqlFunctions { } } + public static ByteString stringToBinary(String s, Charset charset) { + if (s == null) { + return null; + } else { + return new ByteString(s.getBytes(charset)); + } + } + /** Helper for CAST(... AS VARBINARY(maxLength)). */ public static ByteString truncate(ByteString s, int maxLength) { if (s == null) { @@ -5218,7 +5332,7 @@ public class SqlFunctions { } /** Support the ARRAY_REPEAT function. */ - public static @Nullable List<Object> repeat(Object element, Object count) { + public static @Nullable List<Object> arrayRepeat(Object element, Object count) { if (count == null) { return null; } @@ -5250,6 +5364,10 @@ public class SqlFunctions { + "and not exceeds the allowed limit."); } + if (posInt == -1) { + // This means "append to the array" + posInt = baseArray.length + 1; + } boolean usePositivePos = posInt > 0; if (usePositivePos) { @@ -5278,7 +5396,10 @@ public class SqlFunctions { return Arrays.asList(newArray); } else { - int posIndex = posInt; + // 1-based index. + // The behavior of this function was changed in Spark 3.4.0. + // https://issues.apache.org/jira/browse/SPARK-44840 + int posIndex = posInt + 1; boolean newPosExtendsArrayLeft = baseArray.length + posIndex < 0; @@ -5366,6 +5487,22 @@ public class SqlFunctions { return list; } + /** Support the EXISTS(list, function1) function. */ + public static @Nullable Boolean exists(List list, Function1<Object, Boolean> function1) { + return nullableExists(list, function1); + } + + /** Support the EXISTS(list, predicate1) function. */ + public static Boolean exists(List list, Predicate1 predicate1) { + for (Object element : list) { + boolean ret = predicate1.apply(element); + if (ret) { + return true; + } + } + return false; + } + /** Support the MAP_CONCAT function. */ public static Map mapConcat(Map... maps) { final Map result = new LinkedHashMap(); @@ -5392,6 +5529,11 @@ public class SqlFunctions { return new ArrayList<>(map.values()); } + /** Support the MAP_CONTAINS_KEY function. */ + public static Boolean mapContainsKey(Map map, Object key) { + return map.containsKey(key); + } + /** Support the MAP_FROM_ARRAYS function. */ public static Map mapFromArrays(List keysArray, List valuesArray) { if (keysArray.size() != valuesArray.size()) { diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlCall.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlCall.java index 28b4eb1621b..9e1471376d0 100755 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlCall.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlCall.java @@ -205,7 +205,7 @@ public abstract class SqlCall extends SqlNode { * Returns a string describing the actual argument types of a call, e.g. "SUBSTR(VARCHAR(12), * NUMBER(3,2), INTEGER)". */ - protected String getCallSignature(SqlValidator validator, @Nullable SqlValidatorScope scope) { + public String getCallSignature(SqlValidator validator, @Nullable SqlValidatorScope scope) { List<String> signatureList = new ArrayList<>(); for (final SqlNode operand : getOperandList()) { final RelDataType argType = diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java index 60fa7489329..b1c3777fde7 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlGroupedWindowFunction.java @@ -16,7 +16,6 @@ */ package org.apache.calcite.sql; -import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.sql.type.ExplicitReturnTypeInference; @@ -26,9 +25,12 @@ import org.apache.calcite.sql.type.SqlOperandTypeInference; import org.apache.calcite.sql.type.SqlReturnTypeInference; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.calcite.sql.validate.SqlMonotonicity; +import org.checkerframework.checker.nullness.qual.Nullable; import java.util.List; +import static com.google.common.base.Preconditions.checkArgument; + /** * SQL function that computes keys by which rows can be partitioned and aggregated. * @@ -46,12 +48,12 @@ import java.util.List; */ public class SqlGroupedWindowFunction extends SqlFunction { - /** The grouped function, if this an auxiliary function; null otherwise. */ - public final SqlGroupedWindowFunction groupFunction; - private final WindowStartEndReturnTypeInference windowStartEndInf = new WindowStartEndReturnTypeInference(); + /** The grouped function, if this an auxiliary function; null otherwise. */ + public final @Nullable SqlGroupedWindowFunction groupFunction; + /** * Creates a SqlGroupedWindowFunction. * @@ -67,29 +69,22 @@ public class SqlGroupedWindowFunction extends SqlFunction { public SqlGroupedWindowFunction( String name, SqlKind kind, - SqlGroupedWindowFunction groupFunction, + @Nullable SqlGroupedWindowFunction groupFunction, SqlReturnTypeInference returnTypeInference, - SqlOperandTypeInference operandTypeInference, - SqlOperandTypeChecker operandTypeChecker, + @Nullable SqlOperandTypeInference operandTypeInference, + @Nullable SqlOperandTypeChecker operandTypeChecker, SqlFunctionCategory category) { super(name, kind, returnTypeInference, operandTypeInference, operandTypeChecker, category); this.groupFunction = groupFunction; - Preconditions.checkArgument(groupFunction == null || groupFunction.groupFunction == null); + checkArgument(groupFunction == null || groupFunction.groupFunction == null); } - /** - * Creates a SqlGroupedWindowFunction. - * - * @param name Function name - * @param kind Kind - * @param groupFunction Group function, if this is an auxiliary; null, if this is a group - * function - */ + @Deprecated // to be removed before 2.0 public SqlGroupedWindowFunction( String name, SqlKind kind, - SqlGroupedWindowFunction groupFunction, - SqlOperandTypeChecker operandTypeChecker) { + @Nullable SqlGroupedWindowFunction groupFunction, + @Nullable SqlOperandTypeChecker operandTypeChecker) { this( name, kind, @@ -100,6 +95,21 @@ public class SqlGroupedWindowFunction extends SqlFunction { SqlFunctionCategory.SYSTEM); } + @Deprecated // to be removed before 2.0 + public SqlGroupedWindowFunction( + SqlKind kind, + @Nullable SqlGroupedWindowFunction groupFunction, + @Nullable SqlOperandTypeChecker operandTypeChecker) { + this( + kind.name(), + kind, + groupFunction, + ReturnTypes.ARG0, + null, + operandTypeChecker, + SqlFunctionCategory.SYSTEM); + } + /** * Creates an auxiliary function from this grouped window function. * diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlIntervalQualifier.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlIntervalQualifier.java index 1d2fc542994..7676b30e0d3 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlIntervalQualifier.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlIntervalQualifier.java @@ -883,7 +883,7 @@ public class SqlIntervalQualifier extends SqlNode { // Note: must check two patterns, since fractional second is optional final int fractionalSecondPrecision = getFractionalSecondPrecision(typeSystem); String intervalPatternWithFracSec = - "(\\d+) (\\d{1,2}):(\\d{1,2}):(\\d{1,2})\\.(\\d{1," + "(\\d+) (\\d{1,2}):(\\d{1,2}):(\\d{1,2})\\.(\\d{0," + fractionalSecondPrecision + "})"; String intervalPatternWithoutFracSec = "(\\d+) (\\d{1,2}):(\\d{1,2}):(\\d{1,2})"; @@ -1032,7 +1032,7 @@ public class SqlIntervalQualifier extends SqlNode { // Note: must check two patterns, since fractional second is optional final int fractionalSecondPrecision = getFractionalSecondPrecision(typeSystem); String intervalPatternWithFracSec = - "(\\d+):(\\d{1,2}):(\\d{1,2})\\.(\\d{1," + fractionalSecondPrecision + "})"; + "(\\d+):(\\d{1,2}):(\\d{1,2})\\.(\\d{0," + fractionalSecondPrecision + "})"; String intervalPatternWithoutFracSec = "(\\d+):(\\d{1,2}):(\\d{1,2})"; Matcher m = Pattern.compile(intervalPatternWithFracSec).matcher(value); @@ -1134,7 +1134,7 @@ public class SqlIntervalQualifier extends SqlNode { // Note: must check two patterns, since fractional second is optional final int fractionalSecondPrecision = getFractionalSecondPrecision(typeSystem); String intervalPatternWithFracSec = - "(\\d+):(\\d{1,2})\\.(\\d{1," + fractionalSecondPrecision + "})"; + "(\\d+):(\\d{1,2})\\.(\\d{0," + fractionalSecondPrecision + "})"; String intervalPatternWithoutFracSec = "(\\d+):(\\d{1,2})"; Matcher m = Pattern.compile(intervalPatternWithFracSec).matcher(value); @@ -1195,7 +1195,7 @@ public class SqlIntervalQualifier extends SqlNode { // e.g. 'SS' or 'SS.SSS' // Note: must check two patterns, since fractional second is optional final int fractionalSecondPrecision = getFractionalSecondPrecision(typeSystem); - String intervalPatternWithFracSec = "(\\d+)\\.(\\d{1," + fractionalSecondPrecision + "})"; + String intervalPatternWithFracSec = "(\\d+)\\.(\\d{0," + fractionalSecondPrecision + "})"; String intervalPatternWithoutFracSec = "(\\d+)"; Matcher m = Pattern.compile(intervalPatternWithFracSec).matcher(value); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlUtil.java index a85bc71e6c9..feeb06fd232 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlUtil.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlUtil.java @@ -77,7 +77,7 @@ import static org.apache.calcite.util.Static.RESOURCE; /** * Flink modification because of CALCITE-7027 * - * <p>Lines 825 ~ 831, should be removed after upgrading to 1.40. + * <p>Lines 827 ~ 833, should be removed after upgrading to 1.40. */ public abstract class SqlUtil { // ~ Constants -------------------------------------------------------------- diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java index feca7c52a8f..ca6277dd205 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlCastFunction.java @@ -32,12 +32,14 @@ import org.apache.calcite.sql.SqlFunction; import org.apache.calcite.sql.SqlFunctionCategory; 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.SqlOperandCountRange; import org.apache.calcite.sql.SqlOperatorBinding; import org.apache.calcite.sql.SqlSyntax; import org.apache.calcite.sql.SqlUtil; import org.apache.calcite.sql.SqlWriter; +import org.apache.calcite.sql.parser.SqlParserPos; import org.apache.calcite.sql.type.FlinkSqlTypeMappingRule; import org.apache.calcite.sql.type.InferTypes; import org.apache.calcite.sql.type.SqlOperandCountRanges; @@ -117,7 +119,7 @@ public class SqlCastFunction extends SqlFunction { static SqlReturnTypeInference returnTypeInference(boolean safe) { return opBinding -> { - assert opBinding.getOperandCount() == 2; + assert opBinding.getOperandCount() <= 3; final RelDataType ret = deriveType( opBinding.getTypeFactory(), @@ -130,7 +132,7 @@ public class SqlCastFunction extends SqlFunction { SqlNode operand0 = callBinding.operand(0); // dynamic parameters and null constants need their types assigned - // to them using the type they are casted to. + // to them using the type they are cast to. if (SqlUtil.isNullLiteral(operand0, false) || operand0 instanceof SqlDynamicParam) { callBinding.getValidator().setValidatedNodeType(operand0, ret); } @@ -209,13 +211,13 @@ public class SqlCastFunction extends SqlFunction { @Override public String getSignatureTemplate(final int operandsCount) { - assert operandsCount == 2; - return "{0}({1} AS {2})"; + assert operandsCount <= 3; + return "{0}({1} AS {2} [FORMAT {3}])"; } @Override public SqlOperandCountRange getOperandCountRange() { - return SqlOperandCountRanges.of(2); + return SqlOperandCountRanges.between(2, 3); } /** @@ -226,6 +228,11 @@ public class SqlCastFunction extends SqlFunction { public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) { final SqlNode left = callBinding.operand(0); final SqlNode right = callBinding.operand(1); + final SqlLiteral format = + callBinding.getOperandCount() > 2 + ? (SqlLiteral) callBinding.operand(2) + : SqlLiteral.createNull(SqlParserPos.ZERO); + if (SqlUtil.isNullLiteral(left, false) || left instanceof SqlDynamicParam) { return true; } @@ -250,7 +257,10 @@ public class SqlCastFunction extends SqlFunction { } return false; } - return true; + + // Validate format argument is string type if included + return SqlUtil.isNullLiteral(format, false) + || SqlLiteral.valueMatchesType(format.getValue(), SqlTypeName.CHAR); } private boolean canCastFrom(RelDataType toType, RelDataType fromType) { @@ -291,7 +301,7 @@ public class SqlCastFunction extends SqlFunction { @Override public void unparse(SqlWriter writer, SqlCall call, int leftPrec, int rightPrec) { - assert call.operandCount() == 2; + assert call.operandCount() <= 3; final SqlWriter.Frame frame = writer.startFunCall(getName()); call.operand(0).unparse(writer, 0, 0); writer.sep("AS"); @@ -299,6 +309,10 @@ public class SqlCastFunction extends SqlFunction { writer.sep("INTERVAL"); } call.operand(1).unparse(writer, 0, 0); + if (call.getOperandList().size() > 2) { + writer.sep("FORMAT"); + call.operand(2).unparse(writer, 0, 0); + } writer.endFunCall(frame); } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlMapValueConstructor.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlMapValueConstructor.java index 66953a902cc..33b2711cf09 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlMapValueConstructor.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlMapValueConstructor.java @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.calcite.sql.fun; import org.apache.calcite.rel.type.RelDataType; @@ -25,12 +24,13 @@ import org.apache.calcite.sql.SqlOperatorBinding; import org.apache.calcite.sql.type.SqlTypeUtil; import org.apache.calcite.sql.validate.SqlValidatorUtil; import org.apache.calcite.util.Pair; -import org.apache.calcite.util.Static; import org.apache.calcite.util.Util; import org.checkerframework.checker.nullness.qual.Nullable; import java.util.List; -import java.util.Objects; + +import static java.util.Objects.requireNonNull; +import static org.apache.calcite.util.Static.RESOURCE; /** * Flink modifications @@ -44,34 +44,41 @@ public class SqlMapValueConstructor extends SqlMultisetValueConstructor { // END FLINK MODIFICATION } + @SuppressWarnings("argument.type.incompatible") + @Override public RelDataType inferReturnType(SqlOperatorBinding opBinding) { - Pair<RelDataType, RelDataType> type = + Pair<@Nullable RelDataType, @Nullable RelDataType> type = getComponentTypes(opBinding.getTypeFactory(), opBinding.collectOperandTypes()); + + // explicit cast elements to component type if they are not same SqlValidatorUtil.adjustTypeForMapConstructor(type, opBinding); + return SqlTypeUtil.createMapType( opBinding.getTypeFactory(), - (RelDataType) Objects.requireNonNull(type.left, "inferred key type"), - (RelDataType) Objects.requireNonNull(type.right, "inferred value type"), + requireNonNull(type.left, "inferred key type"), + requireNonNull(type.right, "inferred value type"), false); } + @Override public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) { - List<RelDataType> argTypes = SqlTypeUtil.deriveType(callBinding, callBinding.operands()); + final List<RelDataType> argTypes = + SqlTypeUtil.deriveType(callBinding, callBinding.operands()); if (argTypes.size() == 0) { - throw callBinding.newValidationError(Static.RESOURCE.mapRequiresTwoOrMoreArgs()); - } else if (argTypes.size() % 2 > 0) { - throw callBinding.newValidationError(Static.RESOURCE.mapRequiresEvenArgCount()); - } else { - Pair<RelDataType, RelDataType> componentType = - getComponentTypes(callBinding.getTypeFactory(), argTypes); - if (null != componentType.left && null != componentType.right) { - return true; - } else if (throwOnFailure) { - throw callBinding.newValidationError(Static.RESOURCE.needSameTypeParameter()); - } else { - return false; + throw callBinding.newValidationError(RESOURCE.mapRequiresTwoOrMoreArgs()); + } + if (argTypes.size() % 2 > 0) { + throw callBinding.newValidationError(RESOURCE.mapRequiresEvenArgCount()); + } + final Pair<@Nullable RelDataType, @Nullable RelDataType> componentType = + getComponentTypes(callBinding.getTypeFactory(), argTypes); + if (null == componentType.left || null == componentType.right) { + if (throwOnFailure) { + throw callBinding.newValidationError(RESOURCE.needSameTypeParameter()); } + return false; } + return true; } private static Pair<@Nullable RelDataType, @Nullable RelDataType> getComponentTypes( diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeAssignmentRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeAssignmentRule.java index 464e490eb76..75380a252d9 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeAssignmentRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeAssignmentRule.java @@ -101,6 +101,7 @@ public class SqlTypeAssignmentRule implements SqlTypeMappingRule { rule.add(SqlTypeName.BIGINT); rule.add(SqlTypeName.DECIMAL); rule.add(SqlTypeName.FLOAT); + rule.add(SqlTypeName.DOUBLE); rules.add(SqlTypeName.FLOAT, rule); // REAL (32 bit floating point) is assignable from... @@ -111,6 +112,7 @@ public class SqlTypeAssignmentRule implements SqlTypeMappingRule { rule.add(SqlTypeName.BIGINT); rule.add(SqlTypeName.DECIMAL); rule.add(SqlTypeName.FLOAT); + rule.add(SqlTypeName.DOUBLE); rule.add(SqlTypeName.REAL); rules.add(SqlTypeName.REAL, rule); @@ -176,6 +178,9 @@ public class SqlTypeAssignmentRule implements SqlTypeMappingRule { SqlTypeName.TIME_WITH_LOCAL_TIME_ZONE, EnumSet.of(SqlTypeName.TIME_WITH_LOCAL_TIME_ZONE)); + // TIME WITH TIME ZONE is assignable from... + rules.add(SqlTypeName.TIME_TZ, EnumSet.of(SqlTypeName.TIME_TZ)); + // TIMESTAMP is assignable from ... rules.add(SqlTypeName.TIMESTAMP, EnumSet.of(SqlTypeName.TIMESTAMP)); @@ -184,6 +189,9 @@ public class SqlTypeAssignmentRule implements SqlTypeMappingRule { SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE, EnumSet.of(SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE)); + // TIMESTAMP WITH TIME ZONE is assignable from... + rules.add(SqlTypeName.TIMESTAMP_TZ, EnumSet.of(SqlTypeName.TIMESTAMP_TZ)); + // GEOMETRY is assignable from ... rule.clear(); rule.add(SqlTypeName.GEOMETRY); @@ -213,6 +221,7 @@ public class SqlTypeAssignmentRule implements SqlTypeMappingRule { rule.add(SqlTypeName.BIGINT); rule.add(SqlTypeName.DECIMAL); rule.add(SqlTypeName.FLOAT); + rule.add(SqlTypeName.DOUBLE); rule.add(SqlTypeName.REAL); rule.add(SqlTypeName.TIME); rule.add(SqlTypeName.DATE); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java index 14ed16d2e9f..3fa27ea4016 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java @@ -122,6 +122,11 @@ public class SqlTypeFactoryImpl extends RelDataTypeFactoryImpl { return canonize(newType); } + @Override + public RelDataType createFunctionSqlType(RelDataType parameterType, RelDataType returnType) { + return canonize(new FunctionSqlType(parameterType, returnType)); + } + @Override public RelDataType createMeasureType(RelDataType valueType) { MeasureSqlType newType = MeasureSqlType.create(valueType); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java index d66dbecebb5..0ec5a90a5ba 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java @@ -74,8 +74,8 @@ import static org.apache.calcite.util.Static.RESOURCE; * <p>FLINK modifications are at lines * * <ol> - * <li>We should use ExtendedSqlRowTypeNameSpec for rows: Lines 1102-1106 - * <li>Should be removed after fixing CALCITE-7062: Lines 1126-1128 + * <li>We should use ExtendedSqlRowTypeNameSpec for rows: Lines 1084-1087 + * <li>Should be removed after fixing CALCITE-7062: Lines 1108-1110 * </ol> */ public abstract class SqlTypeUtil { @@ -474,24 +474,6 @@ public abstract class SqlTypeUtil { return typeName == SqlTypeName.DECIMAL; } - /** Returns whether a type is DOUBLE. */ - public static boolean isDouble(RelDataType type) { - SqlTypeName typeName = type.getSqlTypeName(); - if (typeName == null) { - return false; - } - return typeName == SqlTypeName.DOUBLE; - } - - /** Returns whether a type is BIGINT. */ - public static boolean isBigint(RelDataType type) { - SqlTypeName typeName = type.getSqlTypeName(); - if (typeName == null) { - return false; - } - return typeName == SqlTypeName.BIGINT; - } - /** Returns whether a type is numeric with exact precision. */ public static boolean isExactNumeric(RelDataType type) { SqlTypeName typeName = type.getSqlTypeName(); diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java index 6ef58067ced..1aa7d5b8054 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java @@ -20,7 +20,6 @@ import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.planner.calcite.FlinkSqlCallBinding; import org.apache.flink.table.planner.functions.sql.ml.SqlVectorSearchTableFunction; -import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; @@ -70,6 +69,7 @@ import org.apache.calcite.sql.SqlIntervalLiteral; import org.apache.calcite.sql.SqlIntervalQualifier; import org.apache.calcite.sql.SqlJoin; import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlLambda; import org.apache.calcite.sql.SqlLiteral; import org.apache.calcite.sql.SqlMatchRecognize; import org.apache.calcite.sql.SqlMerge; @@ -137,6 +137,7 @@ import java.util.AbstractList; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; +import java.util.BitSet; import java.util.Calendar; import java.util.Collection; import java.util.Collections; @@ -145,18 +146,26 @@ import java.util.GregorianCalendar; import java.util.HashMap; import java.util.HashSet; import java.util.IdentityHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.TreeSet; import java.util.function.BiFunction; +import java.util.function.Consumer; import java.util.function.Supplier; import java.util.function.UnaryOperator; import java.util.stream.Collectors; +import java.util.stream.StreamSupport; +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkState; +import static java.util.Collections.emptyList; import static java.util.Objects.requireNonNull; import static org.apache.calcite.linq4j.Nullness.castNonNull; +import static org.apache.calcite.linq4j.Ord.forEach; import static org.apache.calcite.sql.SqlUtil.stripAs; import static org.apache.calcite.sql.type.NonNullableAccessors.getCharset; import static org.apache.calcite.sql.type.NonNullableAccessors.getCollation; @@ -169,35 +178,35 @@ import static org.apache.calcite.util.Util.first; * Default implementation of {@link SqlValidator}, the class was copied over because of * CALCITE-4554. * - * <p>Lines 219 ~ 222, Flink improves error message for functions without appropriate arguments in + * <p>Lines 228 ~ 231, Flink improves error message for functions without appropriate arguments in * handleUnresolvedFunction. * - * <p>Lines 1287 ~ 1289, CALCITE-7217, should be removed after upgrading Calcite to 1.41.0. + * <p>Lines 1317 ~ 1319, CALCITE-7217, should be removed after upgrading Calcite to 1.41.0. * - * <p>Lines 2048 ~ 2062, Flink improves error message for functions without appropriate arguments in + * <p>Lines 2078 ~ 2092, Flink improves error message for functions without appropriate arguments in * handleUnresolvedFunction at {@link SqlValidatorImpl#handleUnresolvedFunction}. * - * <p>Lines 2475 ~ 2477, CALCITE-7471 should be removed after upgrading Calcite to 1.42.0. + * <p>Lines 2505 ~ 2507, CALCITE-7471 should be removed after upgrading Calcite to 1.42.0. * - * <p>Lines 2590 ~ 2609, CALCITE-7217, CALCITE-7312 should be removed after upgrading Calcite to + * <p>Lines 2620 ~ 2639, CALCITE-7217, CALCITE-7312 should be removed after upgrading Calcite to * 1.42.0. * - * <p>Line 2640 ~2658, set the correct scope for VECTOR_SEARCH. + * <p>Line 2670 ~2688, set the correct scope for VECTOR_SEARCH. * - * <p>Lines 3937 ~ 3941, 6612 ~ 6618 Flink improves Optimize the retrieval of sub-operands in + * <p>Lines 4070 ~ 4074, 6758 ~ 6764 Flink improves Optimize the retrieval of sub-operands in * SqlCall when using NamedParameters at {@link SqlValidatorImpl#checkRollUp}. * - * <p>Lines 5357 ~ 5363, FLINK-24352 Add null check for temporal table check on SqlSnapshot. + * <p>Lines 5490 ~ 5496, FLINK-24352 Add null check for temporal table check on SqlSnapshot. * - * <p>Lines 5784-5786, CALCITE-7466 should be removed after upgrading Calcite to 1.42.0. + * <p>Lines 5930-5932, CALCITE-7466 should be removed after upgrading Calcite to 1.42.0. * - * <p>Lines 5840-5842, CALCITE-7470 should be removed after upgrading Calcite to 1.42.0. + * <p>Lines 5986-5988, CALCITE-7470 should be removed after upgrading Calcite to 1.42.0. * - * <p>Lines 7267-7290, CALCITE-7486 should be removed after upgrading Calcite to 1.42.0. + * <p>Lines 7414-7437, CALCITE-7486 should be removed after upgrading Calcite to 1.42.0. * - * <p>Lines 7337-7354, CALCITE-7486 should be removed after upgrading Calcite to 1.42.0. + * <p>Lines 7484-7501, CALCITE-7486 should be removed after upgrading Calcite to 1.42.0. * - * <p>Lines 7399-7407, CALCITE-7486 should be removed after upgrading Calcite to 1.42.0. + * <p>Lines 7546-7554, CALCITE-7486 should be removed after upgrading Calcite to 1.42.0. */ public class SqlValidatorImpl implements SqlValidatorWithHints { // ~ Static fields/initializers --------------------------------------------- @@ -780,7 +789,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { private static int calculatePermuteOffset(List<SqlNode> selectItems) { for (int i = 0; i < selectItems.size(); i++) { SqlNode selectItem = selectItems.get(i); - SqlNode col = SqlUtil.stripAs(selectItem); + SqlNode col = stripAs(selectItem); if (col.getKind() == SqlKind.IDENTIFIER && selectItem.getKind() != SqlKind.AS) { return i; } @@ -1134,6 +1143,22 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { SqlNode node = namespace.getNode(); if (node != null) { setValidatedNodeType(node, namespace.getType()); + + if (node == top) { + // A top-level namespace must not return any must-filter fields. + // A non-top-level namespace (e.g. a subquery) may return must-filter + // fields; these are neutralized if the consuming query filters on them. + final ImmutableBitSet mustFilterFields = namespace.getMustFilterFields(); + if (!mustFilterFields.isEmpty()) { + // Set of field names, sorted alphabetically for determinism. + Set<String> fieldNameSet = + StreamSupport.stream(mustFilterFields.spliterator(), false) + .map(namespace.getRowType().getFieldNames()::get) + .collect(Collectors.toCollection(TreeSet::new)); + throw newValidationError( + node, RESOURCE.mustFilterFieldsMissing(fieldNameSet.toString())); + } + } } } @@ -1202,6 +1227,11 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { return getScopeOrThrow(node); } + @Override + public SqlValidatorScope getLambdaScope(SqlLambda node) { + return getScopeOrThrow(node); + } + @Override public SqlValidatorScope getJoinScope(SqlNode node) { return requireNonNull(scopes.get(stripAs(node)), () -> "scope for " + node); @@ -1880,7 +1910,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { } final SqlNode original = originalExprs.get(node); if (original != null && original != node) { - return getValidatedNodeType(original); + return getValidatedNodeTypeIfKnown(original); } if (node instanceof SqlIdentifier) { return getCatalogReader().getNamedType((SqlIdentifier) node); @@ -1955,7 +1985,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { return ns.getType(); } type = deriveTypeImpl(scope, expr); - Preconditions.checkArgument(type != null, "SqlValidator.deriveTypeInternal returned null"); + requireNonNull(type, "SqlValidator.deriveTypeInternal returned null"); setValidatedNodeType(expr, type); return type; } @@ -2823,14 +2853,14 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { * @param node Query node * @param alias Name of this query within its parent. Must be specified if usingScope != null */ - private void registerQuery( + protected void registerQuery( SqlValidatorScope parentScope, @Nullable SqlValidatorScope usingScope, SqlNode node, SqlNode enclosingNode, @Nullable String alias, boolean forceNullable) { - Preconditions.checkArgument(usingScope == null || alias != null); + checkArgument(usingScope == null || alias != null); registerQuery(parentScope, usingScope, node, enclosingNode, alias, forceNullable, true); } @@ -2854,7 +2884,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { boolean checkUpdate) { requireNonNull(node, "node"); requireNonNull(enclosingNode, "enclosingNode"); - Preconditions.checkArgument(usingScope == null || alias != null); + checkArgument(usingScope == null || alias != null); SqlCall call; List<SqlNode> operands; @@ -2895,7 +2925,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { } } - // If this is an aggregating query, the SELECT list and HAVING + // If this is an aggregate query, the SELECT list and HAVING // clause use a different scope, where you can only reference // columns which are in the GROUP BY clause. SqlValidatorScope aggScope = selectScope; @@ -2925,7 +2955,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { registerSubQueries(orderScope, orderList); if (!isAggregate(select)) { - // Since this is not an aggregating query, + // Since this is not an aggregate query, // there cannot be any aggregates in the ORDER BY clause. SqlNode agg = aggFinder.findAgg(orderList); if (agg != null) { @@ -2949,6 +2979,19 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { registerSetop(parentScope, usingScope, node, enclosingNode, alias, forceNullable); break; + case LAMBDA: + call = (SqlCall) node; + SqlLambdaScope lambdaScope = new SqlLambdaScope(parentScope, (SqlLambda) call); + scopes.put(call, lambdaScope); + final LambdaNamespace lambdaNamespace = + new LambdaNamespace(this, (SqlLambda) call, node); + registerNamespace(usingScope, alias, lambdaNamespace, forceNullable); + operands = call.getOperandList(); + for (int i = 0; i < operands.size(); i++) { + registerOperandSubQueries(parentScope, call, i); + } + break; + case WITH: registerWith( parentScope, @@ -3247,6 +3290,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { return; } if (node.getKind().belongsTo(SqlKind.QUERY) + || node.getKind() == SqlKind.LAMBDA || node.getKind() == SqlKind.MULTISET_QUERY_CONSTRUCTOR || node.getKind() == SqlKind.MULTISET_VALUE_CONSTRUCTOR) { registerQuery(parentScope, null, node, node, null, false); @@ -3334,6 +3378,8 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { break; case DOUBLE: + case FLOAT: + case REAL: validateLiteralAsDouble(literal); break; @@ -3594,7 +3640,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { // Validate condition. switch (conditionType) { case NONE: - Preconditions.checkArgument(join.getCondition() == null); + checkArgument(join.getCondition() == null); break; case ON: final SqlNode condition = expand(getCondition(join), joinScope); @@ -3607,7 +3653,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { List<SqlIdentifier> list = (List) getCondition(join); // Parser ensures that using clause is not empty. - Preconditions.checkArgument(!list.isEmpty(), "Empty USING clause"); + checkArgument(!list.isEmpty(), "Empty USING clause"); for (SqlIdentifier id : list) { validateCommonJoinColumn(id, left, right, scope, natural); } @@ -3714,7 +3760,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { } private RelDataType checkAndDeriveDataType(SqlIdentifier id, SqlNode node) { - Preconditions.checkArgument(id.names.size() == 1); + checkArgument(id.names.size() == 1); String name = id.names.get(0); SqlNameMatcher nameMatcher = getCatalogReader().nameMatcher(); RelDataType rowType = getNamespaceOrThrow(node).getRowType(); @@ -3731,7 +3777,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { */ private RelDataType validateCommonInputJoinColumn( SqlIdentifier id, SqlNode leftOrRight, SqlValidatorScope scope, boolean natural) { - Preconditions.checkArgument(id.names.size() == 1); + checkArgument(id.names.size() == 1); final String name = id.names.get(0); final SqlValidatorNamespace namespace = getNamespaceOrThrow(leftOrRight); final RelDataType rowType = namespace.getRowType(); @@ -3763,7 +3809,8 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { * @param targetRowType Desired row type, must not be null, may be the data type 'unknown'. */ protected void validateSelect(SqlSelect select, RelDataType targetRowType) { - assert targetRowType != null; + requireNonNull(targetRowType, "targetRowType"); + // Namespace is either a select namespace or a wrapper around one. final SelectNamespace ns = getNamespaceOrThrow(select).unwrap(SelectNamespace.class); @@ -3815,12 +3862,13 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { RESOURCE.fromAliasDuplicate(child.name)); } - if (select.getFrom() == null) { + final SqlNode from = select.getFrom(); + if (from == null) { if (this.config.conformance().isFromRequired()) { throw newValidationError(select, RESOURCE.selectMissingFrom()); } } else { - validateFrom(select.getFrom(), fromType, fromScope); + validateFrom(from, fromType, fromScope); } validateWhereClause(select); @@ -3836,12 +3884,78 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { final RelDataType rowType = validateSelectList(selectItems, select, targetRowType); ns.setType(rowType); + // Deduce which columns must be filtered. + ns.mustFilterFields = ImmutableBitSet.of(); + if (from != null) { + final Set<SqlQualified> qualifieds = new LinkedHashSet<>(); + for (ScopeChild child : fromScope.children) { + final List<String> fieldNames = child.namespace.getRowType().getFieldNames(); + child.namespace + .getMustFilterFields() + .forEachInt( + i -> + qualifieds.add( + SqlQualified.create( + fromScope, + 1, + child.namespace, + new SqlIdentifier( + ImmutableList.of( + child.name, + fieldNames.get(i)), + SqlParserPos.ZERO)))); + } + if (!qualifieds.isEmpty()) { + if (select.getWhere() != null) { + forEachQualified(select.getWhere(), getWhereScope(select), qualifieds::remove); + } + if (select.getHaving() != null) { + forEachQualified( + select.getHaving(), getHavingScope(select), qualifieds::remove); + } + + // Each of the must-filter fields identified must be returned as a + // SELECT item, which is then flagged as must-filter. + final BitSet mustFilterFields = new BitSet(); + final List<SqlNode> expandedSelectItems = + requireNonNull(fromScope.getExpandedSelectList(), "expandedSelectList"); + forEach( + expandedSelectItems, + (selectItem, i) -> { + selectItem = stripAs(selectItem); + if (selectItem instanceof SqlIdentifier) { + SqlQualified qualified = + fromScope.fullyQualify((SqlIdentifier) selectItem); + if (qualifieds.remove(qualified)) { + // SELECT item #i referenced a must-filter column that was not + // filtered in the WHERE or HAVING. It becomes a must-filter + // column for our consumer. + mustFilterFields.set(i); + } + } + }); + + // If there are must-filter fields that are not in the SELECT clause, + // this is an error. + if (!qualifieds.isEmpty()) { + throw newValidationError( + select, + RESOURCE.mustFilterFieldsMissing( + qualifieds.stream() + .map(q -> q.suffix().get(0)) + .collect(Collectors.toCollection(TreeSet::new)) + .toString())); + } + ns.mustFilterFields = ImmutableBitSet.fromBitSet(mustFilterFields); + } + } + // Validate ORDER BY after we have set ns.rowType because in some // dialects you can refer to columns of the select list, e.g. // "SELECT empno AS x FROM emp ORDER BY x" validateOrderList(select); - if (shouldCheckForRollUp(select.getFrom())) { + if (shouldCheckForRollUp(from)) { checkRollUpInSelectList(select); checkRollUp(null, select, select.getWhere(), getWhereScope(select)); checkRollUp(null, select, select.getHaving(), getHavingScope(select)); @@ -3851,6 +3965,23 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { } } + /** + * For each identifier in an expression, resolves it to a qualified name and calls the provided + * action. + */ + private static void forEachQualified( + SqlNode node, SqlValidatorScope scope, Consumer<SqlQualified> consumer) { + node.accept( + new SqlBasicVisitor<Void>() { + @Override + public Void visit(SqlIdentifier id) { + final SqlQualified qualified = scope.fullyQualify(id); + consumer.accept(qualified); + return null; + } + }); + } + private void checkRollUpInSelectList(SqlSelect select) { SqlValidatorScope scope = getSelectScope(select); for (SqlNode item : SqlNonNullableAccessors.getSelectList(select)) { @@ -3933,6 +4064,8 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { // only need to check operand[0] for CONVERT or TRANSLATE SqlNode child = ((SqlCall) stripDot).getOperandList().get(0); checkRollUp(parent, current, child, scope, contextClause); + } else if (stripDot.getKind() == SqlKind.LAMBDA) { + // do not need to check lambda } else { // ----- FLINK MODIFICATION BEGIN ----- SqlCall call = (SqlCall) stripDot; @@ -4599,7 +4732,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { private static boolean isReturnBooleanType(RelDataType relDataType) { if (relDataType instanceof RelRecordType) { RelRecordType recordType = (RelRecordType) relDataType; - Preconditions.checkState( + checkState( recordType.getFieldList().size() == 1, "sub-query as condition must return only one column"); RelDataTypeField recordField = recordType.getFieldList().get(0); @@ -4707,10 +4840,10 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { } } - // Unless 'naked measures' are enabled, a non-aggregating query cannot - // reference measure columns. (An aggregating query can use them as + // Unless 'naked measures' are enabled, a non-aggregate query cannot + // reference measure columns. (An aggregate query can use them as // argument to the AGGREGATE function.) - if (!config.nakedMeasures() + if (!config.nakedMeasuresInNonAggregateQuery() && !(scope instanceof AggregatingScope) && scope.isMeasureRef(expr)) { throw newValidationError(expr, RESOURCE.measureMustBeInAggregateQuery()); @@ -5609,6 +5742,18 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { inWindow = false; } + @Override + public void validateLambda(SqlLambda lambdaExpr) { + final SqlLambdaScope scope = (SqlLambdaScope) scopes.get(lambdaExpr); + requireNonNull(scope, "scope"); + final LambdaNamespace ns = getNamespaceOrThrow(lambdaExpr).unwrap(LambdaNamespace.class); + + deriveType(scope, lambdaExpr.getExpression()); + RelDataType type = deriveTypeImpl(scope, lambdaExpr); + setValidatedNodeType(lambdaExpr, type); + ns.setType(type); + } + @Override public void validateMatchRecognize(SqlCall call) { final SqlMatchRecognize matchRecognize = (SqlMatchRecognize) call; @@ -5752,12 +5897,13 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { } }); - final RelDataType rowType = typeBuilder.build(); - if (matchRecognize.getMeasureList().size() == 0) { - ns.setType(getNamespaceOrThrow(matchRecognize.getTableRef()).getRowType()); + final RelDataType rowType; + if (matchRecognize.getMeasureList().isEmpty()) { + rowType = getNamespaceOrThrow(matchRecognize.getTableRef()).getRowType(); } else { - ns.setType(rowType); + rowType = typeBuilder.build(); } + ns.setType(rowType); } private PairList<String, RelDataType> validateMeasure( @@ -6009,7 +6155,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { // Gather the name and type of each measure. final PairList<String, RelDataType> measureNameTypes = PairList.of(); - Ord.forEach( + forEach( unpivot.measureList, (measure, i) -> { final String measureName = ((SqlIdentifier) measure).getSimple(); @@ -6048,7 +6194,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { // ('CLERK', 'ANALYST'), namely VARCHAR(7). The derived type of 'deptno' is // the type of values (10, 20), namely INTEGER. final PairList<String, RelDataType> axisNameTypes = PairList.of(); - Ord.forEach( + forEach( unpivot.axisList, (axis, i) -> { final String axisName = ((SqlIdentifier) axis).getSimple(); @@ -6163,7 +6309,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { final SqlAggFunction op = (SqlAggFunction) aggCall.getOperator(); switch (op.requiresGroupOrder()) { case MANDATORY: - if (orderList == null || orderList.size() == 0) { + if (orderList == null || orderList.isEmpty()) { throw newValidationError( aggCall, RESOURCE.aggregateMissingWithinGroupClause(op.getName())); } @@ -6177,7 +6323,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { } break; case FORBIDDEN: - if (orderList != null && orderList.size() != 0) { + if (orderList != null && !orderList.isEmpty()) { throw newValidationError( aggCall, RESOURCE.withinGroupClauseIllegalInAggregate(op.getName())); } @@ -6324,7 +6470,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { @Override public List<@Nullable List<String>> getFieldOrigins(SqlNode sqlQuery) { if (sqlQuery instanceof SqlExplain) { - return Collections.emptyList(); + return emptyList(); } final RelDataType rowType = getValidatedNodeType(sqlQuery); final int fieldCount = rowType.getFieldCount(); @@ -6568,7 +6714,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { @Override public Void visit(SqlIdentifier id) { - Preconditions.checkArgument(id.isSimple()); + checkArgument(id.isSimple()); scope.addPatternVar(id.getSimple()); return null; } @@ -6765,6 +6911,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { case CURRENT_VALUE: case NEXT_VALUE: case WITH: + case LAMBDA: return call; default: break; @@ -7028,7 +7175,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints { // SQL ordinals are 1-based, but Sort's are 0-based int ordinal = intValue - 1; - return SqlUtil.stripAs( + return stripAs( SqlNonNullableAccessors.getSelectList(select).get(ordinal)); } break; diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java index 4f00abc858b..da6359b6977 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java @@ -95,6 +95,7 @@ import org.apache.calcite.rex.RexDynamicParam; import org.apache.calcite.rex.RexFieldAccess; import org.apache.calcite.rex.RexFieldCollation; import org.apache.calcite.rex.RexInputRef; +import org.apache.calcite.rex.RexLambdaRef; import org.apache.calcite.rex.RexLiteral; import org.apache.calcite.rex.RexNode; import org.apache.calcite.rex.RexOver; @@ -131,6 +132,7 @@ import org.apache.calcite.sql.SqlInsert; import org.apache.calcite.sql.SqlIntervalQualifier; import org.apache.calcite.sql.SqlJoin; import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlLambda; import org.apache.calcite.sql.SqlLiteral; import org.apache.calcite.sql.SqlMatchRecognize; import org.apache.calcite.sql.SqlMerge; @@ -174,6 +176,7 @@ import org.apache.calcite.sql.validate.ListScope; import org.apache.calcite.sql.validate.MatchRecognizeScope; import org.apache.calcite.sql.validate.ParameterScope; import org.apache.calcite.sql.validate.SelectScope; +import org.apache.calcite.sql.validate.SqlLambdaScope; import org.apache.calcite.sql.validate.SqlMonotonicity; import org.apache.calcite.sql.validate.SqlNameMatcher; import org.apache.calcite.sql.validate.SqlQualified; @@ -245,21 +248,21 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * <p>FLINK modifications are at lines * * <ol> - * <li>Added in FLINK-29081, FLINK-28682, FLINK-33395: Lines 688 ~ 705 - * <li>Added in FLINK-24024: Lines 1455 ~ 1461 - * <li>Added in FLINK-24024: Lines 1475 ~ 1514 - * <li>Added in FLINK-37269: Lines 2252 ~ 2274 - * <li>Added in FLINK-28682: Lines 2385 ~ 2402 - * <li>Added in FLINK-28682: Lines 2439 ~ 2467 - * <li>Added in FLINK-32474: Lines 2524 ~ 2526 - * <li>Added in FLINK-32474: Lines 2530 ~ 2532 - * <li>Added in FLINK-32474: Lines 2548 ~ 2550 - * <li>Added in CALCITE-7217: Lines 2589 ~ 2597, should be dropped with upgrade to Calcite 1.41.0 - * <li>Added in FLINK-32474: Lines 2972 ~ 2984 - * <li>Added in FLINK-32474: Lines 3085 ~ 3119 - * <li>Added in FLINK-38720: Lines 4579 ~ 4585 - * <li>Added in FLINK-38720: Lines 4591 ~ 4607 - * <li>Added in FLINK-34312: Lines 5971 ~ 5982 + * <li>Added in FLINK-29081, FLINK-28682, FLINK-33395: Lines 691 ~ 708 + * <li>Added in FLINK-24024: Lines 1458 ~ 1464 + * <li>Added in FLINK-24024: Lines 1478 ~ 1517 + * <li>Added in FLINK-37269: Lines 2255 ~ 2277 + * <li>Added in FLINK-28682: Lines 2419 ~ 2436 + * <li>Added in FLINK-28682: Lines 2473 ~ 2501 + * <li>Added in FLINK-32474: Lines 2558 ~ 2560 + * <li>Added in FLINK-32474: Lines 2564 ~ 2566 + * <li>Added in FLINK-32474: Lines 2582 ~ 2584 + * <li>Added in CALCITE-7217: Lines 2623 ~ 2631, should be dropped with upgrade to Calcite 1.41.0 + * <li>Added in FLINK-32474: Lines 3006 ~ 3018 + * <li>Added in FLINK-32474: Lines 3119 ~ 3153 + * <li>Added in FLINK-38720: Lines 4613 ~ 4617 + * <li>Added in FLINK-38720: Lines 4623 ~ 4639 + * <li>Added in FLINK-34312: Lines 6027 ~ 6038 * </ol> * * <p>In official extension point (i.e. {@link #convertExtendedExpression(SqlNode, Blackboard)}): @@ -2275,6 +2278,37 @@ public class SqlToRelConverter { return null; } + /** + * Converts a lambda expression to a RexNode. + * + * @param bb Blackboard + * @param node Lambda expression + * @return Relational expression + */ + private RexNode convertLambda(Blackboard bb, SqlNode node) { + final SqlLambda call = (SqlLambda) node; + final SqlLambdaScope scope = (SqlLambdaScope) validator().getLambdaScope(call); + + final Map<String, RexNode> nameToNodeMap = new HashMap<>(); + final List<RexLambdaRef> parameters = new ArrayList<>(scope.getParameterTypes().size()); + final Map<String, RelDataType> parameterTypes = scope.getParameterTypes(); + + int i = 0; + for (SqlNode p : call.getParameters()) { + final String name = p.toString(); + final RexLambdaRef parameter = + new RexLambdaRef(i, name, requireNonNull(parameterTypes.get(name))); + parameters.add(parameter); + nameToNodeMap.put(name, parameter); + i++; + } + + final Blackboard lambdaBb = createBlackboard(scope, nameToNodeMap, false); + lambdaBb.setRoot(castNonNull(bb.inputs)); + final RexNode expr = lambdaBb.convertExpression(call.getExpression()); + return rexBuilder.makeLambdaCall(expr, parameters); + } + private RexNode convertOver(Blackboard bb, SqlNode node) { SqlCall call = (SqlCall) node; SqlCall aggCall = call.operand(0); @@ -5616,23 +5650,28 @@ public class SqlToRelConverter { builder.add(convertExpression(node)); } final ImmutableList<RexNode> list = builder.build(); + RelNode rel = root.rel; + // Fix the correlation namespaces and de-duplicate the correlation + // variables. + CorrelationUse correlationUse = getCorrelationUse(this, root.rel); + if (correlationUse != null) { + rel = correlationUse.r; + } + switch (kind) { case IN: - return RexSubQuery.in(root.rel, list); + return RexSubQuery.in(rel, list); case NOT_IN: return rexBuilder.makeCall( - SqlStdOperatorTable.NOT, - RexSubQuery.in(root.rel, list)); + SqlStdOperatorTable.NOT, RexSubQuery.in(rel, list)); case SOME: return RexSubQuery.some( - root.rel, - list, - (SqlQuantifyOperator) call.getOperator()); + rel, list, (SqlQuantifyOperator) call.getOperator()); case ALL: return rexBuilder.makeCall( SqlStdOperatorTable.NOT, RexSubQuery.some( - root.rel, + rel, list, negate( (SqlQuantifyOperator) @@ -5648,6 +5687,13 @@ public class SqlToRelConverter { query = Iterables.getOnlyElement(call.getOperandList()); root = convertQueryRecursive(query, false, null); RelNode rel = root.rel; + // Fix the correlation namespaces and de-duplicate the correlation + // variables. + CorrelationUse correlationUse = getCorrelationUse(this, root.rel); + if (correlationUse != null) { + rel = correlationUse.r; + } + while (rel instanceof Project || rel instanceof Sort && ((Sort) rel).fetch == null @@ -5666,12 +5712,21 @@ public class SqlToRelConverter { call = (SqlCall) expr; query = Iterables.getOnlyElement(call.getOperandList()); root = convertQueryRecursive(query, false, null); - return RexSubQuery.scalar(root.rel); + rel = root.rel; + // Fix the correlation namespaces and de-duplicate the correlation + // variables. + correlationUse = getCorrelationUse(this, root.rel); + if (correlationUse != null) { + rel = correlationUse.r; + } + return RexSubQuery.scalar(rel); case ARRAY_QUERY_CONSTRUCTOR: call = (SqlCall) expr; query = Iterables.getOnlyElement(call.getOperandList()); - root = convertQueryRecursive(query, false, null); + // let top=true to make the query be top-level query, + // then ORDER BY will be reserved. + root = convertQueryRecursive(query, true, null); return RexSubQuery.array(root.rel); case MAP_QUERY_CONSTRUCTOR: @@ -5745,6 +5800,9 @@ public class SqlToRelConverter { case OVER: return convertOver(this, expr); + case LAMBDA: + return convertLambda(this, expr); + default: // fall through } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/StandardConvertletTable.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/StandardConvertletTable.java index 8d4cea40d79..08f9c4e1bb8 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/StandardConvertletTable.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/StandardConvertletTable.java @@ -108,7 +108,7 @@ import static org.apache.calcite.util.Util.first; * <p>FLINK modifications are at lines * * <ol> - * <li>Added in Flink-35216: Lines 833 ~ 879 + * <li>Added in Flink-35216: Lines 843 ~ 889 * </ol> */ public class StandardConvertletTable extends ReflectiveConvertletTable { @@ -123,6 +123,7 @@ public class StandardConvertletTable extends ReflectiveConvertletTable { // Register aliases (operators which have a different name but // identical behavior to other operators). + addAlias(SqlLibraryOperators.LEN, SqlStdOperatorTable.CHAR_LENGTH); addAlias(SqlLibraryOperators.LENGTH, SqlStdOperatorTable.CHAR_LENGTH); addAlias(SqlStdOperatorTable.CHARACTER_LENGTH, SqlStdOperatorTable.CHAR_LENGTH); addAlias(SqlStdOperatorTable.IS_UNKNOWN, SqlStdOperatorTable.IS_NULL); @@ -131,6 +132,10 @@ public class StandardConvertletTable extends ReflectiveConvertletTable { addAlias(SqlStdOperatorTable.PERCENT_REMAINDER, SqlStdOperatorTable.MOD); addAlias(SqlLibraryOperators.IFNULL, SqlLibraryOperators.NVL); addAlias(SqlLibraryOperators.REGEXP_SUBSTR, SqlLibraryOperators.REGEXP_EXTRACT); + addAlias(SqlLibraryOperators.ENDSWITH, SqlLibraryOperators.ENDS_WITH); + addAlias(SqlLibraryOperators.STARTSWITH, SqlLibraryOperators.STARTS_WITH); + addAlias(SqlLibraryOperators.BITAND_AGG, SqlStdOperatorTable.BIT_AND); + addAlias(SqlLibraryOperators.BITOR_AGG, SqlStdOperatorTable.BIT_OR); // Register convertlets for specific objects. registerOp(SqlStdOperatorTable.CAST, this::convertCast); @@ -636,7 +641,13 @@ public class StandardConvertletTable extends ReflectiveConvertletTable { final boolean safe = kind == SqlKind.SAFE_CAST; final SqlNode left = call.operand(0); final SqlNode right = call.operand(1); + final SqlLiteral format = + call.getOperandList().size() > 2 + ? call.operand(2) + : SqlLiteral.createNull(SqlParserPos.ZERO); final RexBuilder rexBuilder = cx.getRexBuilder(); + final RexNode arg = cx.convertExpression(left); + final RexLiteral formatArg = (RexLiteral) cx.convertLiteral(format); if (right instanceof SqlIntervalQualifier) { final SqlIntervalQualifier intervalQualifier = (SqlIntervalQualifier) right; if (left instanceof SqlIntervalLiteral) { @@ -659,7 +670,6 @@ public class StandardConvertletTable extends ReflectiveConvertletTable { return castToValidatedType(call, value, validator, rexBuilder, safe); } - final RexNode arg = cx.convertExpression(left); final SqlDataTypeSpec dataType = (SqlDataTypeSpec) right; RelDataType type = SqlCastFunction.deriveType( @@ -695,7 +705,7 @@ public class StandardConvertletTable extends ReflectiveConvertletTable { type = typeFactory.createTypeWithNullability(type, isn); } } - return rexBuilder.makeCast(type, arg, safe, safe); + return rexBuilder.makeCast(type, arg, safe, safe, formatArg); } protected RexNode convertFloorCeil(SqlRexContext cx, SqlCall call) { diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalOverAggregate.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalOverAggregate.java index 429d24f1f25..d5766793d4c 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalOverAggregate.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalOverAggregate.java @@ -74,6 +74,12 @@ public class FlinkLogicalOverAggregate extends Window implements FlinkLogicalRel super(cluster, traitSet, input, constants, rowType, groups); } + @Override + public Window copy(List<RexLiteral> windowConstants) { + return new FlinkLogicalOverAggregate( + getCluster(), traitSet, input, windowConstants, rowType, groups); + } + @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) { return copy(traitSet, inputs, rowType, groups); diff --git a/flink-table/flink-table-planner/src/main/resources/META-INF/NOTICE b/flink-table/flink-table-planner/src/main/resources/META-INF/NOTICE index 25c814938fd..ffedf681769 100644 --- a/flink-table/flink-table-planner/src/main/resources/META-INF/NOTICE +++ b/flink-table/flink-table-planner/src/main/resources/META-INF/NOTICE @@ -8,9 +8,9 @@ This project bundles the following dependencies under the Apache Software Licens - com.google.guava:guava:33.4.0-jre - com.google.guava:failureaccess:1.0.2 -- org.apache.calcite:calcite-core:1.36.0 -- org.apache.calcite:calcite-linq4j:1.36.0 -- org.apache.calcite.avatica:avatica-core:1.23.0 +- org.apache.calcite:calcite-core:1.37.0 +- org.apache.calcite:calcite-linq4j:1.37.0 +- org.apache.calcite.avatica:avatica-core:1.25.0 - org.apache.commons:commons-math3:3.6.1 - org.apache.commons:commons-text:1.10.0 - commons-codec:commons-codec:1.15 diff --git a/flink-table/flink-table-planner/src/main/resources/META-INF/services/org.codehaus.commons.compiler.ICompilerFactory b/flink-table/flink-table-planner/src/main/resources/META-INF/services/org.codehaus.commons.compiler.ICompilerFactory new file mode 100644 index 00000000000..70af69e0f7d --- /dev/null +++ b/flink-table/flink-table-planner/src/main/resources/META-INF/services/org.codehaus.commons.compiler.ICompilerFactory @@ -0,0 +1,16 @@ +# 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. + +org.codehaus.janino.CompilerFactory diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala index c8023b3ac49..0fc9db1e1c1 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ExprCodeGenerator.scala @@ -598,6 +598,14 @@ class ExprCodeGenerator( override def visitPatternFieldRef(fieldRef: RexPatternFieldRef): GeneratedExpression = throw new CodeGenException("Pattern field references are not supported yet.") + override def visitLambda(var1: RexLambda): GeneratedExpression = { + throw new CodeGenException("Lambdas are not supported yet.") + } + + override def visitLambdaRef(var1: RexLambdaRef): GeneratedExpression = { + throw new CodeGenException("Lambda references are not supported yet.") + } + // ---------------------------------------------------------------------------------------- private def generateCallExpression( diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexDefaultVisitor.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexDefaultVisitor.scala index b3d5e10c856..24f7e36cab5 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexDefaultVisitor.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexDefaultVisitor.scala @@ -61,5 +61,9 @@ abstract class RexDefaultVisitor[R] extends RexVisitor[R] { override def visitLiteral(literal: RexLiteral): R = visitNode(literal) + override def visitLambda(lambda: RexLambda): R = visitNode(lambda) + + override def visitLambdaRef(lambdaRef: RexLambdaRef): R = visitNode(lambdaRef) + def visitNode(rexNode: RexNode): R } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala index 6a81f158bfb..4e2d1d66205 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala @@ -582,6 +582,10 @@ class RexNodeToExpressionConverter( override def visitPatternFieldRef(fieldRef: RexPatternFieldRef): Option[ResolvedExpression] = None + override def visitLambda(rexLambda: RexLambda): Option[ResolvedExpression] = None + + override def visitLambdaRef(rexLambdaRef: RexLambdaRef): Option[ResolvedExpression] = None + private def lookupFunction( identifier: UnresolvedIdentifier, operands: Seq[ResolvedExpression], diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml index 61395048aa0..3d3a6f0cc85 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml @@ -531,9 +531,9 @@ LogicalProject(a=[$0], b=[$1]) <![CDATA[ LogicalProject(a=[$0]) +- LogicalFilter(condition=[NOT(EXISTS({ -LogicalFilter(condition=[=($1, $cor0.b)]) +LogicalFilter(condition=[=($1, $cor1.b)]) LogicalTableScan(table=[[default_catalog, default_database, y]]) -}))], variablesSet=[[$cor0]]) +}))], variablesSet=[[$cor1]]) +- LogicalTableScan(table=[[default_catalog, default_database, x]]) ]]> </Resource> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml index 831d380b7af..a63e9c9cf6d 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml @@ -234,9 +234,9 @@ LogicalProject(a=[$0], b=[$1]) <![CDATA[ LogicalProject(a=[$0]) +- LogicalFilter(condition=[EXISTS({ -LogicalFilter(condition=[=($1, $cor0.b)]) +LogicalFilter(condition=[=($1, $cor1.b)]) LogicalTableScan(table=[[default_catalog, default_database, y]]) -})], variablesSet=[[$cor0]]) +})], variablesSet=[[$cor1]]) +- LogicalTableScan(table=[[default_catalog, default_database, x]]) ]]> </Resource> diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala index d2a2c686cf8..8785e5a77f6 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala @@ -1596,9 +1596,6 @@ class SubQuerySemiJoinTest extends SubQueryTestBase { // LogicalFilter lost variablesSet information. util.verifyRelPlanNotExpected(sqlQuery1, "joinType=[semi]") - - val sqlQuery2 = "SELECT MAX(a) FROM x GROUP BY 1 HAVING EXISTS (SELECT 1 FROM y WHERE d < b)" - util.verifyRelPlanNotExpected(sqlQuery2, "joinType=[semi]") } @Test diff --git a/flink-table/flink-table-runtime/src/main/resources/META-INF/NOTICE b/flink-table/flink-table-runtime/src/main/resources/META-INF/NOTICE index ca53ae9237c..9889a10512e 100644 --- a/flink-table/flink-table-runtime/src/main/resources/META-INF/NOTICE +++ b/flink-table/flink-table-runtime/src/main/resources/META-INF/NOTICE @@ -7,5 +7,5 @@ The Apache Software Foundation (http://www.apache.org/). This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt) - com.jayway.jsonpath:json-path:2.7.0 -- org.codehaus.janino:janino:3.1.10 -- org.codehaus.janino:commons-compiler:3.1.10 +- org.codehaus.janino:janino:3.1.11 +- org.codehaus.janino:commons-compiler:3.1.11 diff --git a/flink-table/pom.xml b/flink-table/pom.xml index 1cc2ba7b79f..769a062fd6a 100644 --- a/flink-table/pom.xml +++ b/flink-table/pom.xml @@ -79,11 +79,8 @@ under the License. </dependencyManagement> <properties> - <calcite.version>1.36.0</calcite.version> - <!-- Calcite 1.36.0 depends on 3.1.9, - at the same time minimum 3.1.x Janino version passing Flink tests without WAs is 3.1.10, - more details are in FLINK-27995 --> - <janino.version>3.1.10</janino.version> + <calcite.version>1.37.0</calcite.version> + <janino.version>3.1.11</janino.version> <guava.version>33.4.0-jre</guava.version> <quartz.version>2.3.2</quartz.version> </properties>
