You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2023/04/18 22:15:24 UTC
[flink] branch master updated (1d9f12a9b40 -> 423cdcb99c3)
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 1d9f12a9b40 [FLINK-31823] Make #testWebRescaleFeatureFlagWithReactiveMode work for the AdaptiveScheduler smoke test.
new 0412329dea4 [FLINK-29214][Table API/SQL] Remove usages of deprecated Aggregate#indicator
new 57f8d58614f [hotfix] replace deprecated calcite apis
new 423cdcb99c3 [FLINK-27998][table] Upgrade Calcite from 1.29.0 to 1.30.0
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../delegation/hive/HiveParserCalcitePlanner.java | 11 +-
.../hive/HiveParserRexNodeConverter.java | 3 +-
.../hive/copy/HiveParserSqlCountAggFunction.java | 7 +-
.../hive/copy/HiveParserSqlMinMaxAggFunction.java | 7 +-
.../hive/copy/HiveParserSqlSumAggFunction.java | 7 +-
flink-table/flink-sql-parser/pom.xml | 4 +-
.../flink/sql/parser/FlinkDDLDataTypeTest.java | 8 +-
.../flink/sql/parser/FlinkSqlParserImplTest.java | 7 +-
.../flink/sql/parser/FlinkSqlUnParserTest.java | 15 +-
flink-table/flink-table-planner/pom.xml | 6 +-
.../org/apache/calcite/rel/core/Correlate.java | 296 -----------
.../calcite/rel/logical/LogicalCorrelate.java | 188 -------
.../org/apache/calcite/sql/fun/SqlRowOperator.java | 128 +++++
.../calcite/sql/type/SqlTypeFactoryImpl.java | 554 +++++++++++++++++++++
.../calcite/sql/validate/SqlValidatorImpl.java | 28 +-
.../apache/calcite/sql2rel/SqlToRelConverter.java | 127 +++--
.../java/org/apache/calcite/tools/RelBuilder.java | 8 +-
.../expressions/converter/OverConvertRule.java | 3 +-
.../planner/functions/sql/SqlListAggFunction.java | 4 +-
.../sql/internal/SqlAuxiliaryGroupAggFunction.java | 4 +-
.../nodes/exec/serde/RexNodeJsonDeserializer.java | 17 +-
.../nodes/exec/serde/RexNodeJsonSerializer.java | 5 +-
...FlinkAggregateExpandDistinctAggregatesRule.java | 2 +-
.../logical/FlinkAggregateJoinTransposeRule.java | 1 -
.../rules/logical/FlinkAggregateRemoveRule.java | 4 +-
.../FlinkSemiAntiJoinFilterTransposeRule.java | 4 +-
.../StreamPhysicalPythonGroupAggregateRule.java | 2 +-
...reamPhysicalPythonGroupWindowAggregateRule.java | 2 +-
.../flink/table/planner/typeutils/SymbolUtil.java | 6 +
.../src/main/resources/META-INF/NOTICE | 4 +-
.../planner/calcite/FlinkLogicalRelFactories.scala | 1 +
.../table/planner/calcite/FlinkTypeFactory.scala | 6 +-
.../planner/calcite/PreValidateReWriter.scala | 1 +
.../planner/plan/metadata/FlinkRelMdSize.scala | 6 +-
.../logical/AggregateReduceGroupingRule.scala | 3 +-
.../plan/rules/logical/CalcRankTransposeRule.scala | 2 +-
.../rules/logical/DecomposeGroupingSetsRule.scala | 2 +-
.../rules/logical/FlinkRewriteSubQueryRule.scala | 2 +-
.../rules/logical/FlinkSubQueryRemoveRule.scala | 4 +-
.../logical/LogicalWindowAggregateRuleBase.scala | 3 +-
.../logical/ProjectSemiAntiJoinTransposeRule.scala | 2 +-
.../rules/logical/PruneAggregateCallRule.scala | 3 +-
.../plan/rules/logical/SplitAggregateRule.scala | 4 +-
.../physical/batch/BatchPhysicalHashAggRule.scala | 4 -
.../batch/BatchPhysicalWindowAggregateRule.scala | 2 +-
.../stream/StreamPhysicalGroupAggregateRule.scala | 2 +-
.../StreamPhysicalGroupWindowAggregateRule.scala | 3 +-
.../stream/StreamPhysicalWindowAggregateRule.scala | 2 +-
.../plan/schema/TimeIndicatorRelDataType.scala | 4 +-
.../planner/plan/trait/FlinkRelDistribution.scala | 2 +-
.../flink/table/planner/plan/trait/TraitUtil.scala | 2 +-
.../plan/nodes/exec/stream/CalcJsonPlanTest.java | 14 +
...JsonPlanITCase.java => SargJsonPlanITCase.java} | 28 +-
.../{testSimpleFilter.out => testSarg.out} | 79 +--
.../metadata/AggCallSelectivityEstimatorTest.scala | 2 +
.../planner/plan/utils/FlinkRexUtilTest.scala | 2 +-
.../src/main/resources/META-INF/NOTICE | 2 +-
flink-table/pom.xml | 6 +-
58 files changed, 983 insertions(+), 672 deletions(-)
delete mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Correlate.java
delete mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalCorrelate.java
create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlRowOperator.java
create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java
copy flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/{LimitJsonPlanITCase.java => SargJsonPlanITCase.java} (63%)
copy flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest_jsonplan/{testSimpleFilter.out => testSarg.out} (67%)
[flink] 03/03: [FLINK-27998][table] Upgrade Calcite from 1.29.0 to 1.30.0
Posted by sn...@apache.org.
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 423cdcb99c3f66be435ad2e70d6a15f10a69e252
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Mon Feb 13 23:07:44 2023 +0100
[FLINK-27998][table] Upgrade Calcite from 1.29.0 to 1.30.0
This closes apache/flink#21934
---
flink-table/flink-sql-parser/pom.xml | 4 +-
.../flink/sql/parser/FlinkDDLDataTypeTest.java | 8 +-
.../flink/sql/parser/FlinkSqlParserImplTest.java | 7 +-
.../flink/sql/parser/FlinkSqlUnParserTest.java | 15 +-
flink-table/flink-table-planner/pom.xml | 6 +-
.../org/apache/calcite/rel/core/Correlate.java | 296 -----------
.../calcite/rel/logical/LogicalCorrelate.java | 188 -------
.../org/apache/calcite/sql/fun/SqlRowOperator.java | 128 +++++
.../calcite/sql/type/SqlTypeFactoryImpl.java | 554 +++++++++++++++++++++
.../calcite/sql/validate/SqlValidatorImpl.java | 28 +-
.../apache/calcite/sql2rel/SqlToRelConverter.java | 127 +++--
.../java/org/apache/calcite/tools/RelBuilder.java | 8 +-
.../nodes/exec/serde/RexNodeJsonDeserializer.java | 17 +-
.../nodes/exec/serde/RexNodeJsonSerializer.java | 5 +-
.../flink/table/planner/typeutils/SymbolUtil.java | 6 +
.../src/main/resources/META-INF/NOTICE | 4 +-
.../planner/calcite/FlinkLogicalRelFactories.scala | 1 +
.../table/planner/calcite/FlinkTypeFactory.scala | 6 +-
.../planner/calcite/PreValidateReWriter.scala | 1 +
.../plan/schema/TimeIndicatorRelDataType.scala | 4 +-
.../plan/nodes/exec/stream/CalcJsonPlanTest.java | 14 +
.../stream/jsonplan/SargJsonPlanITCase.java | 46 ++
.../stream/CalcJsonPlanTest_jsonplan/testSarg.out | 157 ++++++
.../src/main/resources/META-INF/NOTICE | 2 +-
flink-table/pom.xml | 6 +-
25 files changed, 1063 insertions(+), 575 deletions(-)
diff --git a/flink-table/flink-sql-parser/pom.xml b/flink-table/flink-sql-parser/pom.xml
index 4412ade6072..4da406fe319 100644
--- a/flink-table/flink-sql-parser/pom.xml
+++ b/flink-table/flink-sql-parser/pom.xml
@@ -62,9 +62,9 @@ under the License.
<version>${calcite.version}</version>
<exclusions>
<!--
- "mvn dependency:tree" as of Calcite 1.29.0:
+ "mvn dependency:tree" as of Calcite 1.30.0:
- [INFO] +- org.apache.calcite:calcite-core:jar:1.29.0:compile
+ [INFO] +- org.apache.calcite:calcite-core:jar:1.30.0:compile
[INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.20.0:compile
[INFO] | \- org.checkerframework:checker-qual:jar:3.10.0:compile
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 1fe0d63ba41..3df21df7abc 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
@@ -547,12 +547,12 @@ class FlinkDDLDataTypeTest {
private final SqlParser.Config parserConfig;
TestFactory() {
- this(DEFAULT_OPTIONS, MockCatalogReaderSimple::new, SqlValidatorUtil::newValidator);
+ this(DEFAULT_OPTIONS, MockCatalogReaderSimple::create, SqlValidatorUtil::newValidator);
}
TestFactory(
Map<String, Object> options,
- SqlTestFactory.MockCatalogReaderFactory catalogReaderFactory,
+ SqlTestFactory.CatalogReaderFactory catalogReaderFactory,
SqlTestFactory.ValidatorFactory validatorFactory) {
this.options = options;
this.validatorFactory = validatorFactory;
@@ -560,7 +560,7 @@ class FlinkDDLDataTypeTest {
createOperatorTable((SqlOperatorTable) options.get("operatorTable"));
this.typeFactory = createTypeFactory((SqlConformance) options.get("conformance"));
Boolean caseSensitive = (Boolean) options.get("caseSensitive");
- this.catalogReader = catalogReaderFactory.create(typeFactory, caseSensitive).init();
+ this.catalogReader = catalogReaderFactory.create(typeFactory, caseSensitive);
this.parserConfig = createParserConfig(options);
}
@@ -589,7 +589,7 @@ class FlinkDDLDataTypeTest {
catalogReader,
typeFactory,
SqlValidator.Config.DEFAULT
- .withSqlConformance(conformance)
+ .withConformance(conformance)
.withTypeCoercionEnabled(enableTypeCoercion));
}
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 b271805399d..86901433b68 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
@@ -24,7 +24,7 @@ import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.parser.SqlParseException;
-import org.apache.calcite.sql.parser.SqlParserImplFactory;
+import org.apache.calcite.sql.parser.SqlParserFixture;
import org.apache.calcite.sql.parser.SqlParserTest;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
@@ -42,9 +42,8 @@ import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
@Execution(CONCURRENT)
class FlinkSqlParserImplTest extends SqlParserTest {
- @Override
- protected SqlParserImplFactory parserImplFactory() {
- return FlinkSqlParserImpl.FACTORY;
+ public SqlParserFixture fixture() {
+ return super.fixture().withConfig(c -> c.withParserFactory(FlinkSqlParserImpl.FACTORY));
}
@Test
diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlUnParserTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlUnParserTest.java
index 09142717241..cced9d2e440 100644
--- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlUnParserTest.java
+++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlUnParserTest.java
@@ -18,6 +18,9 @@
package org.apache.flink.sql.parser;
+import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl;
+
+import org.apache.calcite.sql.parser.SqlParserFixture;
import org.junit.jupiter.api.parallel.Execution;
import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
@@ -34,13 +37,9 @@ class FlinkSqlUnParserTest extends FlinkSqlParserImplTest {
// ~ Methods ----------------------------------------------------------------
- @Override
- protected boolean isUnparserTest() {
- return true;
- }
-
- @Override
- protected Tester getTester() {
- return new UnparsingTesterImpl();
+ public SqlParserFixture fixture() {
+ return super.fixture()
+ .withTester(new UnparsingTesterImpl())
+ .withConfig(c -> c.withParserFactory(FlinkSqlParserImpl.FACTORY));
}
}
diff --git a/flink-table/flink-table-planner/pom.xml b/flink-table/flink-table-planner/pom.xml
index 7bf89eef10e..9cb54e44b80 100644
--- a/flink-table/flink-table-planner/pom.xml
+++ b/flink-table/flink-table-planner/pom.xml
@@ -129,10 +129,10 @@ under the License.
<version>${calcite.version}</version>
<exclusions>
<!--
- "mvn dependency:tree" as of Calcite 1.29.0:
+ "mvn dependency:tree" as of Calcite 1.30.0:
- [INFO] +- org.apache.calcite:calcite-core:jar:1.29.0:compile
- [INFO] | +- org.apache.calcite:calcite-linq4j:jar:1.29.0:compile
+ [INFO] +- org.apache.calcite:calcite-core:jar:1.30.0:compile
+ [INFO] | +- org.apache.calcite:calcite-linq4j:jar:1.30.0:compile
[INFO] | +- com.esri.geometry:esri-geometry-api:jar:2.2.0:compile
[INFO] | +- com.fasterxml.jackson.core:jackson-annotations:jar:2.13.4:compile
[INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.20.0:compile
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Correlate.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Correlate.java
deleted file mode 100644
index 180010eabb5..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Correlate.java
+++ /dev/null
@@ -1,296 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.calcite.rel.core;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptCost;
-import org.apache.calcite.plan.RelOptPlanner;
-import org.apache.calcite.plan.RelOptUtil;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.BiRel;
-import org.apache.calcite.rel.RelInput;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelWriter;
-import org.apache.calcite.rel.hint.Hintable;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.sql.validate.SqlValidatorUtil;
-import org.apache.calcite.util.ImmutableBitSet;
-import org.apache.calcite.util.Litmus;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Set;
-
-import static java.util.Objects.requireNonNull;
-
-/**
- * A relational operator that performs nested-loop joins.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- *
- * <p>It behaves like a kind of {@link org.apache.calcite.rel.core.Join}, but works by setting
- * variables in its environment and restarting its right-hand input.
- *
- * <p>Correlate is not a join since: typical rules should not match Correlate.
- *
- * <p>A Correlate is used to represent a correlated query. One implementation strategy is to
- * de-correlate the expression.
- *
- * <table>
- * <caption>Mapping of physical operations to logical ones</caption>
- * <tr><th>Physical operation</th><th>Logical operation</th></tr>
- * <tr><td>NestedLoops</td><td>Correlate(A, B, regular)</td></tr>
- * <tr><td>NestedLoopsOuter</td><td>Correlate(A, B, outer)</td></tr>
- * <tr><td>NestedLoopsSemi</td><td>Correlate(A, B, semi)</td></tr>
- * <tr><td>NestedLoopsAnti</td><td>Correlate(A, B, anti)</td></tr>
- * <tr><td>HashJoin</td><td>EquiJoin(A, B)</td></tr>
- * <tr><td>HashJoinOuter</td><td>EquiJoin(A, B, outer)</td></tr>
- * <tr><td>HashJoinSemi</td><td>SemiJoin(A, B, semi)</td></tr>
- * <tr><td>HashJoinAnti</td><td>SemiJoin(A, B, anti)</td></tr>
- * </table>
- *
- * @see CorrelationId
- */
-public abstract class Correlate extends BiRel implements Hintable {
- // ~ Instance fields --------------------------------------------------------
-
- protected final CorrelationId correlationId;
- protected final ImmutableBitSet requiredColumns;
- protected final JoinRelType joinType;
- protected final ImmutableList<RelHint> hints;
-
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a Correlate.
- *
- * @param cluster Cluster this relational expression belongs to
- * @param hints Hints for this node
- * @param left Left input relational expression
- * @param right Right input relational expression
- * @param correlationId Variable name for the row of left input
- * @param requiredColumns Set of columns that are used by correlation
- * @param joinType Join type
- */
- @SuppressWarnings("method.invocation.invalid")
- protected Correlate(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- RelNode left,
- RelNode right,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType) {
- super(cluster, traitSet, left, right);
- assert !joinType.generatesNullsOnLeft() : "Correlate has invalid join type " + joinType;
- this.joinType = requireNonNull(joinType, "joinType");
- this.correlationId = requireNonNull(correlationId, "correlationId");
- this.requiredColumns = requireNonNull(requiredColumns, "requiredColumns");
- this.hints = ImmutableList.copyOf(hints);
- }
-
- /**
- * Creates a Correlate.
- *
- * @param cluster Cluster this relational expression belongs to
- * @param left Left input relational expression
- * @param right Right input relational expression
- * @param correlationId Variable name for the row of left input
- * @param requiredColumns Set of columns that are used by correlation
- * @param joinType Join type
- */
- protected Correlate(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelNode left,
- RelNode right,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType) {
- this(
- cluster,
- traitSet,
- Collections.emptyList(),
- left,
- right,
- correlationId,
- requiredColumns,
- joinType);
- }
-
- /**
- * Creates a Correlate by parsing serialized output.
- *
- * @param input Input representation
- */
- protected Correlate(RelInput input) {
- this(
- input.getCluster(),
- input.getTraitSet(),
- input.getInputs().get(0),
- input.getInputs().get(1),
- new CorrelationId(
- requireNonNull((Integer) input.get("correlation"), "correlation")),
- input.getBitSet("requiredColumns"),
- requireNonNull(input.getEnum("joinType", JoinRelType.class), "joinType"));
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public boolean isValid(Litmus litmus, RelNode.Context context) {
- ImmutableBitSet leftColumns = ImmutableBitSet.range(left.getRowType().getFieldCount());
- return super.isValid(litmus, context)
- && litmus.check(
- leftColumns.contains(requiredColumns),
- "Required columns {} not subset of left columns {}",
- requiredColumns,
- leftColumns)
- && RelOptUtil.notContainsCorrelation(left, correlationId, litmus);
- }
-
- @Override
- public Correlate copy(RelTraitSet traitSet, List<RelNode> inputs) {
- assert inputs.size() == 2;
- return copy(
- traitSet, inputs.get(0), inputs.get(1), correlationId, requiredColumns, joinType);
- }
-
- public abstract Correlate copy(
- RelTraitSet traitSet,
- RelNode left,
- RelNode right,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType);
-
- public JoinRelType getJoinType() {
- return joinType;
- }
-
- @Override
- protected RelDataType deriveRowType() {
- switch (joinType) {
- case LEFT:
- case INNER:
- return SqlValidatorUtil.deriveJoinRowType(
- left.getRowType(),
- right.getRowType(),
- joinType,
- getCluster().getTypeFactory(),
- null,
- ImmutableList.of());
- case ANTI:
- case SEMI:
- return left.getRowType();
- default:
- throw new IllegalStateException("Unknown join type " + joinType);
- }
- }
-
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- return super.explainTerms(pw)
- .item("correlation", correlationId)
- .item("joinType", joinType.lowerName)
- .item("requiredColumns", requiredColumns);
- }
-
- /**
- * Returns the correlating expressions.
- *
- * @return correlating expressions
- */
- public CorrelationId getCorrelationId() {
- return correlationId;
- }
-
- @Override
- public String getCorrelVariable() {
- return correlationId.getName();
- }
-
- /**
- * Returns the required columns in left relation required for the correlation in the right.
- *
- * @return columns in left relation required for the correlation in the right
- */
- public ImmutableBitSet getRequiredColumns() {
- return requiredColumns;
- }
-
- @Override
- public Set<CorrelationId> getVariablesSet() {
- return ImmutableSet.of(correlationId);
- }
-
- @Override
- public double estimateRowCount(RelMetadataQuery mq) {
- double leftRowCount = mq.getRowCount(left);
- switch (joinType) {
- case SEMI:
- case ANTI:
- return leftRowCount;
- default:
- return leftRowCount * mq.getRowCount(right);
- }
- }
-
- @Override
- public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
- double rowCount = mq.getRowCount(this);
-
- final double rightRowCount = right.estimateRowCount(mq);
- final double leftRowCount = left.estimateRowCount(mq);
- if (Double.isInfinite(leftRowCount) || Double.isInfinite(rightRowCount)) {
- return planner.getCostFactory().makeInfiniteCost();
- }
-
- Double restartCount = mq.getRowCount(getLeft());
- if (restartCount == null) {
- return planner.getCostFactory().makeInfiniteCost();
- }
- // RelMetadataQuery.getCumulativeCost(getRight()); does not work for
- // RelSubset, so we ask planner to cost-estimate right relation
- RelOptCost rightCost = planner.getCost(getRight(), mq);
- if (rightCost == null) {
- return planner.getCostFactory().makeInfiniteCost();
- }
- RelOptCost rescanCost = rightCost.multiplyBy(Math.max(1.0, restartCount - 1));
-
- return planner.getCostFactory()
- .makeCost(
- rowCount /* generate results */ + leftRowCount /* scan left results */,
- 0,
- 0)
- .plus(rescanCost);
- }
-
- @Override
- public ImmutableList<RelHint> getHints() {
- return hints;
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalCorrelate.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalCorrelate.java
deleted file mode 100644
index fe56ba5b600..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalCorrelate.java
+++ /dev/null
@@ -1,188 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.calcite.rel.logical;
-
-import org.apache.calcite.plan.Convention;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelInput;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelShuttle;
-import org.apache.calcite.rel.core.Correlate;
-import org.apache.calcite.rel.core.CorrelationId;
-import org.apache.calcite.rel.core.JoinRelType;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.util.ImmutableBitSet;
-
-import java.util.Collections;
-import java.util.List;
-
-import static java.util.Objects.requireNonNull;
-
-/**
- * A relational operator that performs nested-loop joins.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- *
- * <p>It behaves like a kind of {@link org.apache.calcite.rel.core.Join}, but works by setting
- * variables in its environment and restarting its right-hand input.
- *
- * <p>A LogicalCorrelate is used to represent a correlated query. One implementation strategy is to
- * de-correlate the expression.
- *
- * @see org.apache.calcite.rel.core.CorrelationId
- */
-public final class LogicalCorrelate extends Correlate {
- // ~ Instance fields --------------------------------------------------------
-
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a LogicalCorrelate.
- *
- * @param cluster cluster this relational expression belongs to
- * @param hints hints for this node
- * @param left left input relational expression
- * @param right right input relational expression
- * @param correlationId variable name for the row of left input
- * @param requiredColumns Required columns
- * @param joinType join type
- */
- public LogicalCorrelate(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- RelNode left,
- RelNode right,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType) {
- super(cluster, traitSet, hints, left, right, correlationId, requiredColumns, joinType);
- }
-
- /**
- * Creates a LogicalCorrelate.
- *
- * @param cluster cluster this relational expression belongs to
- * @param left left input relational expression
- * @param right right input relational expression
- * @param correlationId variable name for the row of left input
- * @param requiredColumns Required columns
- * @param joinType join type
- */
- public LogicalCorrelate(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelNode left,
- RelNode right,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType) {
- this(
- cluster,
- traitSet,
- Collections.emptyList(),
- left,
- right,
- correlationId,
- requiredColumns,
- joinType);
- }
-
- /** Creates a LogicalCorrelate by parsing serialized output. */
- public LogicalCorrelate(RelInput input) {
- this(
- input.getCluster(),
- input.getTraitSet(),
- input.getInputs().get(0),
- input.getInputs().get(1),
- new CorrelationId(
- (Integer) requireNonNull(input.get("correlation"), "correlation")),
- input.getBitSet("requiredColumns"),
- requireNonNull(input.getEnum("joinType", JoinRelType.class), "joinType"));
- }
-
- /** Creates a LogicalCorrelate. */
- public static LogicalCorrelate create(
- RelNode left,
- RelNode right,
- List<RelHint> hints,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType) {
- final RelOptCluster cluster = left.getCluster();
- final RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE);
- return new LogicalCorrelate(
- cluster, traitSet, hints, left, right, correlationId, requiredColumns, joinType);
- }
-
- /** Creates a LogicalCorrelate. */
- public static LogicalCorrelate create(
- RelNode left,
- RelNode right,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType) {
- final RelOptCluster cluster = left.getCluster();
- final RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE);
- return new LogicalCorrelate(
- cluster, traitSet, left, right, correlationId, requiredColumns, joinType);
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public LogicalCorrelate copy(
- RelTraitSet traitSet,
- RelNode left,
- RelNode right,
- CorrelationId correlationId,
- ImmutableBitSet requiredColumns,
- JoinRelType joinType) {
- assert traitSet.containsIfApplicable(Convention.NONE);
- return new LogicalCorrelate(
- getCluster(),
- traitSet,
- hints,
- left,
- right,
- correlationId,
- requiredColumns,
- joinType);
- }
-
- @Override
- public RelNode accept(RelShuttle shuttle) {
- return shuttle.visit(this);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalCorrelate(
- getCluster(),
- traitSet,
- hintList,
- left,
- right,
- correlationId,
- requiredColumns,
- joinType);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlRowOperator.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlRowOperator.java
new file mode 100644
index 00000000000..d861f2af623
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/fun/SqlRowOperator.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.calcite.sql.fun;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlUtil;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.type.InferTypes;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.util.Pair;
+
+import java.util.AbstractList;
+import java.util.Map;
+
+/**
+ * Copied to keep null semantics of table api and sql in sync. At the same time SQL standard says
+ * that the next about `ROW`:
+ *
+ * <ul>
+ * <li>The value of {@code R IS NULL} is:
+ * <ul>
+ * <li>If the value of every field of V is the null value, then True.
+ * <li>Otherwise, False.
+ * </ul>
+ * <li>The value of {@code R IS NOT NULL} is:
+ * <ul>
+ * <li>If the value of no field of V is the null value, then True.
+ * <li>Otherwise, False.
+ * </ul>
+ * </ul>
+ *
+ * <p>Calcite applies that logic since <a
+ * href="https://issues.apache.org/jira/browse/CALCITE-3627">CALCITE-3627</a> (1.30.0+).
+ *
+ * <ul>
+ * <li>Thus, with Calcite 1.30.0+
+ * <ul>
+ * <li>{@code SELECT ROW(CAST(NULL AS INT), CAST(NULL AS INT)) IS NOT NULL; -- returns
+ * FALSE}
+ * <li>{@code SELECT ROW(CAST(NULL AS INT), CAST(NULL AS INT)) IS NULL; -- returns TRUE}.
+ * </ul>
+ * <li>With Flink and Calcite before 1.30.0 (current behavior of this class)
+ * <ul>
+ * <li>{@code SELECT ROW(CAST(NULL AS INT), CAST(NULL AS INT)) IS NOT NULL; -- returns TRUE}
+ * <li>{@code SELECT ROW(CAST(NULL AS INT), CAST(NULL AS INT)) IS NULL; -- returns FALSE}
+ * </ul>
+ * </ul>
+ *
+ * Once Flink applies same logic for both table api and sql, this class should be removed.
+ *
+ * <p>Changed lines
+ *
+ * <ol>
+ * <li>Line 92 ~ 112
+ * </ol>
+ */
+public class SqlRowOperator extends SqlSpecialOperator {
+ // ~ Constructors -----------------------------------------------------------
+
+ public SqlRowOperator(String name) {
+ super(
+ name,
+ SqlKind.ROW,
+ MDX_PRECEDENCE,
+ false,
+ null,
+ InferTypes.RETURN_TYPE,
+ OperandTypes.VARIADIC);
+ }
+
+ // ~ Methods ----------------------------------------------------------------
+
+ @Override
+ public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
+ // ----- FLINK MODIFICATION BEGIN -----
+ // The type of a ROW(e1,e2) expression is a record with the types
+ // {e1type,e2type}. According to the standard, field names are
+ // implementation-defined.
+ return opBinding
+ .getTypeFactory()
+ .createStructType(
+ new AbstractList<Map.Entry<String, RelDataType>>() {
+ @Override
+ public Map.Entry<String, RelDataType> get(int index) {
+ return Pair.of(
+ SqlUtil.deriveAliasFromOrdinal(index),
+ opBinding.getOperandType(index));
+ }
+
+ @Override
+ public int size() {
+ return opBinding.getOperandCount();
+ }
+ });
+ // ----- FLINK MODIFICATION END -----
+ }
+
+ @Override
+ public void unparse(SqlWriter writer, SqlCall call, int leftPrec, int rightPrec) {
+ SqlUtil.unparseFunctionSyntax(this, writer, call, false);
+ }
+
+ // override SqlOperator
+ @Override
+ public boolean requiresDecimalExpansion() {
+ return false;
+ }
+}
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
new file mode 100644
index 00000000000..a55c40475e2
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java
@@ -0,0 +1,554 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.sql.type;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataTypeFamily;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.sql.SqlCollation;
+import org.apache.calcite.sql.SqlIntervalQualifier;
+import org.apache.calcite.util.Util;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.nio.charset.Charset;
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Default implementation {@link SqlTypeFactoryImpl}, the class was copied over because of
+ * FLINK-31350.
+ *
+ * <p>FLINK modifications are at lines
+ *
+ * <ol>
+ * <li>Should be removed after fix of FLINK-31350: Lines 541 ~ 553.
+ * </ol>
+ */
+public class SqlTypeFactoryImpl extends RelDataTypeFactoryImpl {
+ // ~ Constructors -----------------------------------------------------------
+
+ public SqlTypeFactoryImpl(RelDataTypeSystem typeSystem) {
+ super(typeSystem);
+ }
+
+ // ~ Methods ----------------------------------------------------------------
+
+ @Override
+ public RelDataType createSqlType(SqlTypeName typeName) {
+ if (typeName.allowsPrec()) {
+ return createSqlType(typeName, typeSystem.getDefaultPrecision(typeName));
+ }
+ assertBasic(typeName);
+ RelDataType newType = new BasicSqlType(typeSystem, typeName);
+ return canonize(newType);
+ }
+
+ @Override
+ public RelDataType createSqlType(SqlTypeName typeName, int precision) {
+ final int maxPrecision = typeSystem.getMaxPrecision(typeName);
+ if (maxPrecision >= 0 && precision > maxPrecision) {
+ precision = maxPrecision;
+ }
+ if (typeName.allowsScale()) {
+ return createSqlType(typeName, precision, typeName.getDefaultScale());
+ }
+ assertBasic(typeName);
+ assert (precision >= 0) || (precision == RelDataType.PRECISION_NOT_SPECIFIED);
+ // Does not check precision when typeName is SqlTypeName#NULL.
+ RelDataType newType =
+ precision == RelDataType.PRECISION_NOT_SPECIFIED
+ ? new BasicSqlType(typeSystem, typeName)
+ : new BasicSqlType(typeSystem, typeName, precision);
+ newType = SqlTypeUtil.addCharsetAndCollation(newType, this);
+ return canonize(newType);
+ }
+
+ @Override
+ public RelDataType createSqlType(SqlTypeName typeName, int precision, int scale) {
+ assertBasic(typeName);
+ assert (precision >= 0) || (precision == RelDataType.PRECISION_NOT_SPECIFIED);
+ final int maxPrecision = typeSystem.getMaxPrecision(typeName);
+ if (maxPrecision >= 0 && precision > maxPrecision) {
+ precision = maxPrecision;
+ }
+ RelDataType newType = new BasicSqlType(typeSystem, typeName, precision, scale);
+ newType = SqlTypeUtil.addCharsetAndCollation(newType, this);
+ return canonize(newType);
+ }
+
+ @Override
+ public RelDataType createUnknownType() {
+ // FLINK MODIFICATION BEGIN
+ return canonize(new UnknownSqlType(this));
+ // FLINK MODIFICATION END
+ }
+
+ @Override
+ public RelDataType createMultisetType(RelDataType type, long maxCardinality) {
+ assert maxCardinality == -1;
+ RelDataType newType = new MultisetSqlType(type, false);
+ return canonize(newType);
+ }
+
+ @Override
+ public RelDataType createArrayType(RelDataType elementType, long maxCardinality) {
+ assert maxCardinality == -1;
+ ArraySqlType newType = new ArraySqlType(elementType, false);
+ return canonize(newType);
+ }
+
+ @Override
+ public RelDataType createMapType(RelDataType keyType, RelDataType valueType) {
+ MapSqlType newType = new MapSqlType(keyType, valueType, false);
+ return canonize(newType);
+ }
+
+ @Override
+ public RelDataType createSqlIntervalType(SqlIntervalQualifier intervalQualifier) {
+ RelDataType newType = new IntervalSqlType(typeSystem, intervalQualifier, false);
+ return canonize(newType);
+ }
+
+ @Override
+ public RelDataType createTypeWithCharsetAndCollation(
+ RelDataType type, Charset charset, SqlCollation collation) {
+ assert SqlTypeUtil.inCharFamily(type) : type;
+ requireNonNull(charset, "charset");
+ requireNonNull(collation, "collation");
+ RelDataType newType;
+ if (type instanceof BasicSqlType) {
+ BasicSqlType sqlType = (BasicSqlType) type;
+ newType = sqlType.createWithCharsetAndCollation(charset, collation);
+ } else if (type instanceof JavaType) {
+ JavaType javaType = (JavaType) type;
+ newType =
+ new JavaType(
+ javaType.getJavaClass(), javaType.isNullable(), charset, collation);
+ } else {
+ throw Util.needToImplement("need to implement " + type);
+ }
+ return canonize(newType);
+ }
+
+ @Override
+ public @Nullable RelDataType leastRestrictive(List<RelDataType> types) {
+ assert types != null;
+ assert types.size() >= 1;
+
+ RelDataType type0 = types.get(0);
+ if (type0.getSqlTypeName() != null) {
+ RelDataType resultType = leastRestrictiveSqlType(types);
+ if (resultType != null) {
+ return resultType;
+ }
+ return leastRestrictiveByCast(types);
+ }
+
+ return super.leastRestrictive(types);
+ }
+
+ private @Nullable RelDataType leastRestrictiveByCast(List<RelDataType> types) {
+ RelDataType resultType = types.get(0);
+ boolean anyNullable = resultType.isNullable();
+ for (int i = 1; i < types.size(); i++) {
+ RelDataType type = types.get(i);
+ if (type.getSqlTypeName() == SqlTypeName.NULL) {
+ anyNullable = true;
+ continue;
+ }
+
+ if (type.isNullable()) {
+ anyNullable = true;
+ }
+
+ if (SqlTypeUtil.canCastFrom(type, resultType, false)) {
+ resultType = type;
+ } else {
+ if (!SqlTypeUtil.canCastFrom(resultType, type, false)) {
+ return null;
+ }
+ }
+ }
+ if (anyNullable) {
+ return createTypeWithNullability(resultType, true);
+ } else {
+ return resultType;
+ }
+ }
+
+ @Override
+ public RelDataType createTypeWithNullability(final RelDataType type, final boolean nullable) {
+ final RelDataType newType;
+ if (type instanceof BasicSqlType) {
+ newType = ((BasicSqlType) type).createWithNullability(nullable);
+ } else if (type instanceof MapSqlType) {
+ newType = copyMapType(type, nullable);
+ } else if (type instanceof ArraySqlType) {
+ newType = copyArrayType(type, nullable);
+ } else if (type instanceof MultisetSqlType) {
+ newType = copyMultisetType(type, nullable);
+ } else if (type instanceof IntervalSqlType) {
+ newType = copyIntervalType(type, nullable);
+ } else if (type instanceof ObjectSqlType) {
+ newType = copyObjectType(type, nullable);
+ } else {
+ return super.createTypeWithNullability(type, nullable);
+ }
+ return canonize(newType);
+ }
+
+ private static void assertBasic(SqlTypeName typeName) {
+ assert typeName != null;
+ assert typeName != SqlTypeName.MULTISET : "use createMultisetType() instead";
+ assert typeName != SqlTypeName.ARRAY : "use createArrayType() instead";
+ assert typeName != SqlTypeName.MAP : "use createMapType() instead";
+ assert typeName != SqlTypeName.ROW : "use createStructType() instead";
+ assert !SqlTypeName.INTERVAL_TYPES.contains(typeName)
+ : "use createSqlIntervalType() instead";
+ }
+
+ private @Nullable RelDataType leastRestrictiveSqlType(List<RelDataType> types) {
+ RelDataType resultType = null;
+ int nullCount = 0;
+ int nullableCount = 0;
+ int javaCount = 0;
+ int anyCount = 0;
+
+ for (RelDataType type : types) {
+ final SqlTypeName typeName = type.getSqlTypeName();
+ if (typeName == null) {
+ return null;
+ }
+ if (typeName == SqlTypeName.ANY) {
+ anyCount++;
+ }
+ if (type.isNullable()) {
+ ++nullableCount;
+ }
+ if (typeName == SqlTypeName.NULL) {
+ ++nullCount;
+ }
+ if (isJavaType(type)) {
+ ++javaCount;
+ }
+ }
+
+ // if any of the inputs are ANY, the output is ANY
+ if (anyCount > 0) {
+ return createTypeWithNullability(
+ createSqlType(SqlTypeName.ANY), nullCount > 0 || nullableCount > 0);
+ }
+
+ for (int i = 0; i < types.size(); ++i) {
+ RelDataType type = types.get(i);
+ RelDataTypeFamily family = type.getFamily();
+
+ final SqlTypeName typeName = type.getSqlTypeName();
+ if (typeName == SqlTypeName.NULL) {
+ continue;
+ }
+
+ // Convert Java types; for instance, JavaType(int) becomes INTEGER.
+ // Except if all types are either NULL or Java types.
+ if (isJavaType(type) && javaCount + nullCount < types.size()) {
+ final RelDataType originalType = type;
+ type =
+ typeName.allowsPrecScale(true, true)
+ ? createSqlType(typeName, type.getPrecision(), type.getScale())
+ : typeName.allowsPrecScale(true, false)
+ ? createSqlType(typeName, type.getPrecision())
+ : createSqlType(typeName);
+ type = createTypeWithNullability(type, originalType.isNullable());
+ }
+
+ if (resultType == null) {
+ resultType = type;
+ SqlTypeName sqlTypeName = resultType.getSqlTypeName();
+ if (sqlTypeName == SqlTypeName.ROW) {
+ return leastRestrictiveStructuredType(types);
+ }
+ if (sqlTypeName == SqlTypeName.ARRAY || sqlTypeName == SqlTypeName.MULTISET) {
+ return leastRestrictiveArrayMultisetType(types, sqlTypeName);
+ }
+ if (sqlTypeName == SqlTypeName.MAP) {
+ return leastRestrictiveMapType(types, sqlTypeName);
+ }
+ }
+
+ RelDataTypeFamily resultFamily = resultType.getFamily();
+ SqlTypeName resultTypeName = resultType.getSqlTypeName();
+
+ if (resultFamily != family) {
+ return null;
+ }
+ if (SqlTypeUtil.inCharOrBinaryFamilies(type)) {
+ Charset charset1 = type.getCharset();
+ Charset charset2 = resultType.getCharset();
+ SqlCollation collation1 = type.getCollation();
+ SqlCollation collation2 = resultType.getCollation();
+
+ final int precision =
+ SqlTypeUtil.maxPrecision(resultType.getPrecision(), type.getPrecision());
+
+ // If either type is LOB, then result is LOB with no precision.
+ // Otherwise, if either is variable width, result is variable
+ // width. Otherwise, result is fixed width.
+ if (SqlTypeUtil.isLob(resultType)) {
+ resultType = createSqlType(resultType.getSqlTypeName());
+ } else if (SqlTypeUtil.isLob(type)) {
+ resultType = createSqlType(type.getSqlTypeName());
+ } else if (SqlTypeUtil.isBoundedVariableWidth(resultType)) {
+ resultType = createSqlType(resultType.getSqlTypeName(), precision);
+ } else {
+ // this catch-all case covers type variable, and both fixed
+
+ SqlTypeName newTypeName = type.getSqlTypeName();
+
+ if (typeSystem.shouldConvertRaggedUnionTypesToVarying()) {
+ if (resultType.getPrecision() != type.getPrecision()) {
+ if (newTypeName == SqlTypeName.CHAR) {
+ newTypeName = SqlTypeName.VARCHAR;
+ } else if (newTypeName == SqlTypeName.BINARY) {
+ newTypeName = SqlTypeName.VARBINARY;
+ }
+ }
+ }
+
+ resultType = createSqlType(newTypeName, precision);
+ }
+ Charset charset = null;
+ // TODO: refine collation combination rules
+ SqlCollation collation0 =
+ collation1 != null && collation2 != null
+ ? SqlCollation.getCoercibilityDyadicOperator(collation1, collation2)
+ : null;
+ SqlCollation collation = null;
+ if ((charset1 != null) || (charset2 != null)) {
+ if (charset1 == null) {
+ charset = charset2;
+ collation = collation2;
+ } else if (charset2 == null) {
+ charset = charset1;
+ collation = collation1;
+ } else if (charset1.equals(charset2)) {
+ charset = charset1;
+ collation = collation1;
+ } else if (charset1.contains(charset2)) {
+ charset = charset1;
+ collation = collation1;
+ } else {
+ charset = charset2;
+ collation = collation2;
+ }
+ }
+ if (charset != null) {
+ resultType =
+ createTypeWithCharsetAndCollation(
+ resultType,
+ charset,
+ collation0 != null
+ ? collation0
+ : requireNonNull(collation, "collation"));
+ }
+ } else if (SqlTypeUtil.isExactNumeric(type)) {
+ if (SqlTypeUtil.isExactNumeric(resultType)) {
+ // TODO: come up with a cleaner way to support
+ // interval + datetime = datetime
+ if (types.size() > (i + 1)) {
+ RelDataType type1 = types.get(i + 1);
+ if (SqlTypeUtil.isDatetime(type1)) {
+ resultType = type1;
+ return createTypeWithNullability(
+ resultType, nullCount > 0 || nullableCount > 0);
+ }
+ }
+ if (!type.equals(resultType)) {
+ if (!typeName.allowsPrec() && !resultTypeName.allowsPrec()) {
+ // use the bigger primitive
+ if (type.getPrecision() > resultType.getPrecision()) {
+ resultType = type;
+ }
+ } else {
+ // Let the result type have precision (p), scale (s)
+ // and number of whole digits (d) as follows: d =
+ // max(p1 - s1, p2 - s2) s <= max(s1, s2) p = s + d
+
+ int p1 = resultType.getPrecision();
+ int p2 = type.getPrecision();
+ int s1 = resultType.getScale();
+ int s2 = type.getScale();
+ final int maxPrecision = typeSystem.getMaxNumericPrecision();
+ final int maxScale = typeSystem.getMaxNumericScale();
+
+ int dout = Math.max(p1 - s1, p2 - s2);
+ dout = Math.min(dout, maxPrecision);
+
+ int scale = Math.max(s1, s2);
+ scale = Math.min(scale, maxPrecision - dout);
+ scale = Math.min(scale, maxScale);
+
+ int precision = dout + scale;
+ assert precision <= maxPrecision;
+ assert precision > 0
+ || (resultType.getSqlTypeName() == SqlTypeName.DECIMAL
+ && precision == 0
+ && scale == 0);
+
+ resultType = createSqlType(SqlTypeName.DECIMAL, precision, scale);
+ }
+ }
+ } else if (SqlTypeUtil.isApproximateNumeric(resultType)) {
+ // already approximate; promote to double just in case
+ // TODO: only promote when required
+ if (SqlTypeUtil.isDecimal(type)) {
+ // Only promote to double for decimal types
+ resultType = createDoublePrecisionType();
+ }
+ } else {
+ return null;
+ }
+ } else if (SqlTypeUtil.isApproximateNumeric(type)) {
+ if (SqlTypeUtil.isApproximateNumeric(resultType)) {
+ if (type.getPrecision() > resultType.getPrecision()) {
+ resultType = type;
+ }
+ } else if (SqlTypeUtil.isExactNumeric(resultType)) {
+ if (SqlTypeUtil.isDecimal(resultType)) {
+ resultType = createDoublePrecisionType();
+ } else {
+ resultType = type;
+ }
+ } else {
+ return null;
+ }
+ } else if (SqlTypeUtil.isInterval(type)) {
+ // TODO: come up with a cleaner way to support
+ // interval + datetime = datetime
+ if (types.size() > (i + 1)) {
+ RelDataType type1 = types.get(i + 1);
+ if (SqlTypeUtil.isDatetime(type1)) {
+ resultType = type1;
+ return createTypeWithNullability(
+ resultType, nullCount > 0 || nullableCount > 0);
+ }
+ }
+
+ if (!type.equals(resultType)) {
+ // TODO jvs 4-June-2005: This shouldn't be necessary;
+ // move logic into IntervalSqlType.combine
+ Object type1 = resultType;
+ resultType =
+ ((IntervalSqlType) resultType).combine(this, (IntervalSqlType) type);
+ resultType =
+ ((IntervalSqlType) resultType).combine(this, (IntervalSqlType) type1);
+ }
+ } else if (SqlTypeUtil.isDatetime(type)) {
+ // TODO: come up with a cleaner way to support
+ // datetime +/- interval (or integer) = datetime
+ if (types.size() > (i + 1)) {
+ RelDataType type1 = types.get(i + 1);
+ if (SqlTypeUtil.isInterval(type1) || SqlTypeUtil.isIntType(type1)) {
+ resultType = type;
+ return createTypeWithNullability(
+ resultType, nullCount > 0 || nullableCount > 0);
+ }
+ }
+ } else {
+ // TODO: datetime precision details; for now we let
+ // leastRestrictiveByCast handle it
+ return null;
+ }
+ }
+ if (resultType != null && nullableCount > 0) {
+ resultType = createTypeWithNullability(resultType, true);
+ }
+ return resultType;
+ }
+
+ private RelDataType createDoublePrecisionType() {
+ return createSqlType(SqlTypeName.DOUBLE);
+ }
+
+ private RelDataType copyMultisetType(RelDataType type, boolean nullable) {
+ MultisetSqlType mt = (MultisetSqlType) type;
+ RelDataType elementType = copyType(mt.getComponentType());
+ return new MultisetSqlType(elementType, nullable);
+ }
+
+ private RelDataType copyIntervalType(RelDataType type, boolean nullable) {
+ return new IntervalSqlType(
+ typeSystem,
+ requireNonNull(
+ type.getIntervalQualifier(),
+ () -> "type.getIntervalQualifier() for " + type),
+ nullable);
+ }
+
+ private static RelDataType copyObjectType(RelDataType type, boolean nullable) {
+ return new ObjectSqlType(
+ type.getSqlTypeName(),
+ type.getSqlIdentifier(),
+ nullable,
+ type.getFieldList(),
+ type.getComparability());
+ }
+
+ private RelDataType copyArrayType(RelDataType type, boolean nullable) {
+ ArraySqlType at = (ArraySqlType) type;
+ RelDataType elementType = copyType(at.getComponentType());
+ return new ArraySqlType(elementType, nullable);
+ }
+
+ private RelDataType copyMapType(RelDataType type, boolean nullable) {
+ MapSqlType mt = (MapSqlType) type;
+ RelDataType keyType = copyType(mt.getKeyType());
+ RelDataType valueType = copyType(mt.getValueType());
+ return new MapSqlType(keyType, valueType, nullable);
+ }
+
+ @Override
+ protected RelDataType canonize(RelDataType type) {
+ type = super.canonize(type);
+ if (!(type instanceof ObjectSqlType)) {
+ return type;
+ }
+ ObjectSqlType objectType = (ObjectSqlType) type;
+ if (!objectType.isNullable()) {
+ objectType.setFamily(objectType);
+ } else {
+ objectType.setFamily((RelDataTypeFamily) createTypeWithNullability(objectType, false));
+ }
+ return type;
+ }
+
+ // FLINK MODIFICATION BEGIN
+ /** The unknown type. Similar to the NULL type, but is only equal to itself. */
+ static class UnknownSqlType extends BasicSqlType {
+ UnknownSqlType(RelDataTypeFactory typeFactory) {
+ super(typeFactory.getTypeSystem(), SqlTypeName.NULL);
+ }
+
+ @Override
+ protected void generateTypeString(StringBuilder sb, boolean withDetail) {
+ sb.append("UNKNOWN");
+ }
+ }
+ // FLINK MODIFICATION END
+}
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 b237f3db638..d05271f03c1 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
@@ -305,7 +305,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
// ~ Methods ----------------------------------------------------------------
public SqlConformance getConformance() {
- return config.sqlConformance();
+ return config.conformance();
}
@Pure
@@ -539,7 +539,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
final SqlIdentifier identifier = (SqlIdentifier) selectItem;
if (!identifier.isSimple()) {
- if (!validator.config().sqlConformance().allowQualifyingCommonColumn()) {
+ if (!validator.config().conformance().allowQualifyingCommonColumn()) {
validateQualifiedCommonColumn((SqlJoin) from, identifier, scope, validator);
}
return selectItem;
@@ -3420,7 +3420,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
// a NATURAL keyword?
switch (joinType) {
case LEFT_SEMI_JOIN:
- if (!this.config.sqlConformance().isLiberal()) {
+ if (!this.config.conformance().isLiberal()) {
throw newValidationError(
join.getJoinTypeNode(),
RESOURCE.dialectDoesNotSupportFeature("LEFT SEMI JOIN"));
@@ -3551,7 +3551,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
}
if (select.getFrom() == null) {
- if (this.config.sqlConformance().isFromRequired()) {
+ if (this.config.conformance().isFromRequired()) {
throw newValidationError(select, RESOURCE.selectMissingFrom());
}
} else {
@@ -4282,7 +4282,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
return;
}
final AggregatingScope havingScope = (AggregatingScope) getSelectScope(select);
- if (config.sqlConformance().isHavingAlias()) {
+ if (config.conformance().isHavingAlias()) {
SqlNode newExpr = expandGroupByOrHavingExpr(having, havingScope, select, true);
if (having != newExpr) {
having = newExpr;
@@ -4732,7 +4732,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
protected RelDataType getLogicalTargetRowType(RelDataType targetRowType, SqlInsert insert) {
if (insert.getTargetColumnList() == null
- && this.config.sqlConformance().isInsertSubsetColumnsAllowed()) {
+ && this.config.conformance().isInsertSubsetColumnsAllowed()) {
// Target an implicit subset of columns.
final SqlNode source = insert.getSource();
final RelDataType sourceRowType = getNamespaceOrThrow(source).getRowType();
@@ -5034,7 +5034,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
}
SqlCall rowConstructor = (SqlCall) operand;
- if (this.config.sqlConformance().isInsertSubsetColumnsAllowed()
+ if (this.config.conformance().isInsertSubsetColumnsAllowed()
&& targetRowType.isStruct()
&& rowConstructor.operandCount() < targetRowType.getFieldCount()) {
targetRowType =
@@ -5655,7 +5655,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
// What columns from the input are not referenced by a column in the IN
// list?
- final SqlValidatorNamespace inputNs = Objects.requireNonNull(getNamespace(unpivot.query));
+ final SqlValidatorNamespace inputNs = requireNonNull(getNamespace(unpivot.query));
final Set<String> unusedColumnNames = catalogReader.nameMatcher().createSet();
unusedColumnNames.addAll(inputNs.getRowType().getFieldNames());
unusedColumnNames.removeAll(unpivot.usedColumnNames());
@@ -5873,7 +5873,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
if ((call.operandCount() == 0)
&& (operator.getSyntax() == SqlSyntax.FUNCTION_ID)
&& !call.isExpanded()
- && !this.config.sqlConformance().allowNiladicParentheses()) {
+ && !this.config.conformance().allowNiladicParentheses()) {
// For example, "LOCALTIME()" is illegal. (It should be
// "LOCALTIME", which would have been handled as a
// SqlIdentifier.)
@@ -6416,7 +6416,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
// Ordinal markers, e.g. 'select a, b from t order by 2'.
// Only recognize them if they are the whole expression,
// and if the dialect permits.
- if (literal == root && config.sqlConformance().isSortByOrdinal()) {
+ if (literal == root && config.conformance().isSortByOrdinal()) {
switch (literal.getTypeName()) {
case DECIMAL:
case DOUBLE:
@@ -6462,7 +6462,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
@Override
public SqlNode visit(SqlIdentifier id) {
// Aliases, e.g. 'select a as x, b from t order by x'.
- if (id.isSimple() && config.sqlConformance().isSortByAlias()) {
+ if (id.isSimple() && config.conformance().isSortByAlias()) {
String alias = id.getSimple();
final SqlValidatorNamespace selectNs = getNamespaceOrThrow(select);
final RelDataType rowType = selectNs.getRowTypeSansSystemColumns();
@@ -6537,8 +6537,8 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
public @Nullable SqlNode visit(SqlIdentifier id) {
if (id.isSimple()
&& (havingExpr
- ? validator.config().sqlConformance().isHavingAlias()
- : validator.config().sqlConformance().isGroupByAlias())) {
+ ? validator.config().conformance().isHavingAlias()
+ : validator.config().conformance().isGroupByAlias())) {
String name = id.getSimple();
SqlNode expr = null;
final SqlNameMatcher nameMatcher = validator.catalogReader.nameMatcher();
@@ -6579,7 +6579,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
@Override
public @Nullable SqlNode visit(SqlLiteral literal) {
- if (havingExpr || !validator.config().sqlConformance().isGroupByOrdinal()) {
+ if (havingExpr || !validator.config().conformance().isGroupByOrdinal()) {
return super.visit(literal);
}
boolean isOrdinalLiteral = literal == root;
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 49aa96ae8f7..2b0844071c3 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
@@ -220,9 +220,9 @@ import static org.apache.calcite.sql.SqlUtil.stripAs;
*
* <ol>
* <li>Added in FLINK-29081, FLINK-28682: Lines 633 ~ 643
- * <li>Added in FLINK-28682: Lines 2095 ~ 2112
- * <li>Added in FLINK-28682: Lines 2149 ~ 2177
- * <li>Added in FLINK-20873: Lines 5198 ~ 5207
+ * <li>Added in FLINK-28682: Lines 2130 ~ 2147
+ * <li>Added in FLINK-28682: Lines 2184 ~ 2212
+ * <li>Added in FLINK-20873: Lines 5245 ~ 5254
* </ol>
*/
@SuppressWarnings("UnstableApiUsage")
@@ -231,7 +231,7 @@ public class SqlToRelConverter {
// ~ Static fields/initializers ---------------------------------------------
/** Default configuration. */
- private static final Config CONFIG =
+ public static final Config CONFIG =
ImmutableSqlToRelConverter.Config.builder()
.withRelBuilderFactory(RelFactories.LOGICAL_BUILDER)
.withRelBuilderConfigTransform(c -> c.withPushJoinCondition(true))
@@ -248,7 +248,7 @@ public class SqlToRelConverter {
// ~ Instance fields --------------------------------------------------------
- protected final @Nullable SqlValidator validator;
+ public final @Nullable SqlValidator validator;
protected final RexBuilder rexBuilder;
protected final Prepare.CatalogReader catalogReader;
protected final RelOptCluster cluster;
@@ -548,7 +548,7 @@ public class SqlToRelConverter {
*/
public RelNode trimUnusedFields(boolean ordered, RelNode rootRel) {
// Trim fields that are not used by their consumer.
- if (isTrimUnusedFields()) {
+ if (config.isTrimUnusedFields()) {
final RelFieldTrimmer trimmer = newFieldTrimmer();
final List<RelCollation> collations =
rootRel.getTraitSet().getTraits(RelCollationTraitDef.INSTANCE);
@@ -698,6 +698,41 @@ public class SqlToRelConverter {
/** Implementation of {@link #convertSelect(SqlSelect, boolean)}; derived class may override. */
protected void convertSelectImpl(final Blackboard bb, SqlSelect select) {
convertFrom(bb, select.getFrom());
+
+ // We would like to remove ORDER BY clause from an expanded view, except if
+ // it is top-level or affects semantics.
+ //
+ // Top-level example. Given the view definition
+ // CREATE VIEW v AS SELECT * FROM t ORDER BY x
+ // we would retain the view's ORDER BY in
+ // SELECT * FROM v
+ // or
+ // SELECT * FROM v WHERE y = 5
+ // but remove the view's ORDER BY in
+ // SELECT * FROM v ORDER BY z
+ // and
+ // SELECT deptno, COUNT(*) FROM v GROUP BY deptno
+ // because the ORDER BY and GROUP BY mean that the view is not 'top level' in
+ // the query.
+ //
+ // Semantics example. Given the view definition
+ // CREATE VIEW v2 AS SELECT * FROM t ORDER BY x LIMIT 10
+ // we would never remove the ORDER BY, because "ORDER BY ... LIMIT" is about
+ // semantics. It is not a 'pure order'.
+ if (RelOptUtil.isPureOrder(castNonNull(bb.root)) && config.isRemoveSortInSubQuery()) {
+ // Remove the Sort if the view is at the top level. Also remove the Sort
+ // if there are other nodes, which will cause the view to be in the
+ // sub-query.
+ if (!bb.top
+ || validator().isAggregate(select)
+ || select.isDistinct()
+ || select.hasOrderBy()
+ || select.getFetch() != null
+ || select.getOffset() != null) {
+ bb.setRoot(castNonNull(bb.root).getInput(0), true);
+ }
+ }
+
convertWhere(bb, select.getWhere());
final List<SqlNode> orderExprList = new ArrayList<>();
@@ -2299,7 +2334,7 @@ public class SqlToRelConverter {
}
RelNode child = (null != bb.root) ? bb.root : LogicalValues.createOneRow(cluster);
RelNode uncollect;
- if (validator().config().sqlConformance().allowAliasUnnestItems()) {
+ if (validator().config().conformance().allowAliasUnnestItems()) {
uncollect =
relBuilder
.push(child)
@@ -2656,15 +2691,19 @@ public class SqlToRelConverter {
SqlValidatorUtil.getExtendedColumns(validator, validatorTable, extendedColumns);
table = table.extend(extendedFields);
}
- final RelNode tableRel;
// Review Danny 2020-01-13: hacky to construct a new table scan
// in order to apply the hint strategies.
final List<RelHint> hints =
hintStrategies.apply(
SqlUtil.getRelHint(hintStrategies, tableHints),
LogicalTableScan.create(cluster, table, ImmutableList.of()));
- tableRel = toRel(table, hints);
+ final RelNode tableRel = toRel(table, hints);
bb.setRoot(tableRel, true);
+
+ if (RelOptUtil.isPureOrder(castNonNull(bb.root)) && removeSortInSubQuery(bb.top)) {
+ bb.setRoot(castNonNull(bb.root).getInput(0), true);
+ }
+
if (usedDataset[0]) {
bb.setDataset(datasetName);
}
@@ -2813,7 +2852,8 @@ public class SqlToRelConverter {
requiredCols = ImmutableBitSet.fromBitSet(shuttle.varCols).union(p.requiredColumns);
}
- return LogicalCorrelate.create(leftRel, innerRel, p.id, requiredCols, joinType);
+ return LogicalCorrelate.create(
+ leftRel, innerRel, ImmutableList.of(), p.id, requiredCols, joinType);
}
final RelNode node =
@@ -3937,7 +3977,7 @@ public class SqlToRelConverter {
final RelDataType tableRowType = targetTable.getRowType();
SqlNodeList targetColumnList = call.getTargetColumnList();
if (targetColumnList == null) {
- if (validator().config().sqlConformance().isInsertSubsetColumnsAllowed()) {
+ if (validator().config().conformance().isInsertSubsetColumnsAllowed()) {
final RelDataType targetRowType =
typeFactory.createStructType(
tableRowType
@@ -4163,17 +4203,12 @@ public class SqlToRelConverter {
} else {
qualified = SqlQualified.create(null, 1, null, identifier);
}
- final Pair<RexNode, @Nullable Map<String, Integer>> e0 =
- requireNonNull(
- bb.lookupExp(qualified), () -> "no expression found for " + qualified);
+ final Pair<RexNode, @Nullable BiFunction<RexNode, String, RexNode>> e0 =
+ bb.lookupExp(qualified);
RexNode e = e0.left;
for (String name : qualified.suffix()) {
if (e == e0.left && e0.right != null) {
- Integer i =
- requireNonNull(
- e0.right.get(name),
- () -> "e0.right.get(name) produced null for " + name);
- e = rexBuilder.makeFieldAccess(e, i);
+ e = e0.right.apply(e, name);
} else {
final boolean caseSensitive = true; // name already fully-qualified
if (identifier.isStar() && bb.scope instanceof MatchRecognizeScope) {
@@ -4649,7 +4684,9 @@ public class SqlToRelConverter {
mapRootRelToFieldProjection.put(newLeftInput, currentProjection);
}
- setRoot(newLeftInput, false);
+ // if the original root rel is a leaf rel, the new root should be a leaf.
+ // otherwise the field offset will be wrong.
+ setRoot(newLeftInput, leaves.remove(root()) != null);
// right fields appear after the LHS fields.
final int rightOffset =
@@ -4758,13 +4795,13 @@ public class SqlToRelConverter {
}
/**
- * Returns an expression with which to reference a from-list item.
+ * Returns an expression with which to reference a from-list item; throws if not found.
*
- * @param qualified the alias of the from item
- * @return a {@link RexFieldAccess} or {@link RexRangeRef}, or null if not found
+ * @param qualified The alias of the FROM item
+ * @return a {@link RexFieldAccess} or {@link RexRangeRef}, never null
*/
- @Nullable
- Pair<RexNode, @Nullable Map<String, Integer>> lookupExp(SqlQualified qualified) {
+ Pair<RexNode, @Nullable BiFunction<RexNode, String, RexNode>> lookupExp(
+ SqlQualified qualified) {
if (nameToNodeMap != null && qualified.prefixLength == 1) {
RexNode node = nameToNodeMap.get(qualified.identifier.names.get(0));
if (node == null) {
@@ -4779,8 +4816,12 @@ public class SqlToRelConverter {
scope().getValidator().getCatalogReader().nameMatcher();
final SqlValidatorScope.ResolvedImpl resolved = new SqlValidatorScope.ResolvedImpl();
scope().resolve(qualified.prefix(), nameMatcher, false, resolved);
- if (!(resolved.count() == 1)) {
- return null;
+ if (resolved.count() != 1) {
+ throw new AssertionError(
+ "no unique expression found for "
+ + qualified
+ + "; count is "
+ + resolved.count());
}
final SqlValidatorScope.Resolve resolve = resolved.only();
final RelDataType rowType = resolve.rowType();
@@ -4793,18 +4834,16 @@ public class SqlToRelConverter {
if ((inputs != null) && !isParent) {
final LookupContext rels = new LookupContext(this, inputs, systemFieldList.size());
final RexNode node = lookup(resolve.path.steps().get(0).i, rels);
- if (node == null) {
- return null;
- } else {
- final Map<String, Integer> fieldOffsets = new HashMap<>();
- for (RelDataTypeField f : resolve.rowType().getFieldList()) {
- if (!fieldOffsets.containsKey(f.getName())) {
- fieldOffsets.put(f.getName(), f.getIndex());
- }
- }
- final Map<String, Integer> map = ImmutableMap.copyOf(fieldOffsets);
- return Pair.of(node, map);
- }
+ assert node != null;
+ return Pair.of(
+ node,
+ (e, fieldName) -> {
+ final RelDataTypeField field =
+ requireNonNull(
+ rowType.getField(fieldName, true, false),
+ () -> "field " + fieldName);
+ return rexBuilder.makeFieldAccess(e, field.getIndex());
+ });
} else {
// We're referencing a relational expression which has not been
// converted yet. This occurs when from items are correlated,
@@ -4833,7 +4872,15 @@ public class SqlToRelConverter {
offset += c.getRowType().getFieldCount();
}
final RexNode c = rexBuilder.makeCorrel(builder.uniquify().build(), correlId);
- return Pair.of(c, fields.build());
+ final ImmutableMap<String, Integer> fieldMap = fields.build();
+ return Pair.of(
+ c,
+ (e, fieldName) -> {
+ final int j =
+ requireNonNull(
+ fieldMap.get(fieldName), "field " + fieldName);
+ return rexBuilder.makeFieldAccess(e, j);
+ });
}
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/tools/RelBuilder.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/tools/RelBuilder.java
index 98c8d7ce3ca..00e5075e3c5 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/tools/RelBuilder.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/tools/RelBuilder.java
@@ -154,7 +154,7 @@ import static org.apache.calcite.util.Static.RESOURCE;
* <p>FLINK modifications are at lines
*
* <ol>
- * <li>Should be removed after fix of FLINK-29804: Lines 2927 ~ 2930
+ * <li>Should be removed after fix of FLINK-29804: Lines 2928 ~ 2931
* </ol>
*/
@Value.Enclosing
@@ -2879,7 +2879,8 @@ public class RelBuilder {
tableSpool(Spool.Type.LAZY, Spool.Type.LAZY, finder.relOptTable).build();
RelNode seed = tableSpool(Spool.Type.LAZY, Spool.Type.LAZY, finder.relOptTable).build();
RelNode repeatUnion =
- struct.repeatUnionFactory.createRepeatUnion(seed, iterative, all, iterationLimit);
+ struct.repeatUnionFactory.createRepeatUnion(
+ seed, iterative, all, iterationLimit, finder.relOptTable);
return push(repeatUnion);
}
@@ -2962,7 +2963,7 @@ public class RelBuilder {
final ImmutableBitSet requiredColumns = RelOptUtil.correlationColumns(id, right.rel);
join =
struct.correlateFactory.createCorrelate(
- left.rel, right.rel, id, requiredColumns, joinType);
+ left.rel, right.rel, ImmutableList.of(), id, requiredColumns, joinType);
} else {
RelNode join0 =
struct.joinFactory.createJoin(
@@ -3020,6 +3021,7 @@ public class RelBuilder {
struct.correlateFactory.createCorrelate(
left.rel,
right.rel,
+ ImmutableList.of(),
correlationId,
ImmutableBitSet.of(requiredOrdinals),
joinType);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonDeserializer.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonDeserializer.java
index cbb7df05680..c5111219283 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonDeserializer.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonDeserializer.java
@@ -45,6 +45,7 @@ import org.apache.calcite.rel.core.CorrelationId;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUnknownAs;
import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlOperator;
@@ -90,6 +91,7 @@ import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSe
import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSerializer.FIELD_NAME_INTERNAL_NAME;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSerializer.FIELD_NAME_KIND;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSerializer.FIELD_NAME_NAME;
+import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSerializer.FIELD_NAME_NULL_AS;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSerializer.FIELD_NAME_OPERANDS;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSerializer.FIELD_NAME_RANGES;
import static org.apache.flink.table.planner.plan.nodes.exec.serde.RexNodeJsonSerializer.FIELD_NAME_SARG;
@@ -217,9 +219,18 @@ final class RexNodeJsonDeserializer extends StdDeserializer<RexNode> {
builder.add(range);
}
}
- final boolean containsNull = sargNode.required(FIELD_NAME_CONTAINS_NULL).booleanValue();
- return rexBuilder.makeSearchArgumentLiteral(
- Sarg.of(containsNull, builder.build()), relDataType);
+ // TODO: Since 1.18.0 nothing is serialized to FIELD_NAME_CONTAINS_NULL.
+ // This if condition (should be removed in future) is required for backward compatibility
+ // with Flink 1.17.x where FIELD_NAME_CONTAINS_NULL is still used.
+ final JsonNode containsNull = sargNode.get(FIELD_NAME_CONTAINS_NULL);
+ if (containsNull != null) {
+ return rexBuilder.makeSearchArgumentLiteral(
+ Sarg.of(containsNull.booleanValue(), builder.build()), relDataType);
+ }
+ final RexUnknownAs nullAs =
+ serializableToCalcite(
+ RexUnknownAs.class, sargNode.required(FIELD_NAME_NULL_AS).asText());
+ return rexBuilder.makeSearchArgumentLiteral(Sarg.of(nullAs, builder.build()), relDataType);
}
private static @Nullable Object deserializeLiteralValue(
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerializer.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerializer.java
index 5e788723cd3..f0b6bf80fae 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerializer.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/RexNodeJsonSerializer.java
@@ -93,7 +93,9 @@ final class RexNodeJsonSerializer extends StdSerializer<RexNode> {
static final String FIELD_NAME_BOUND_LOWER = "lower";
static final String FIELD_NAME_BOUND_UPPER = "upper";
static final String FIELD_NAME_BOUND_TYPE = "boundType";
+
static final String FIELD_NAME_CONTAINS_NULL = "containsNull";
+ static final String FIELD_NAME_NULL_AS = "nullAs";
// Symbol fields
static final String FIELD_NAME_SYMBOL = "symbol";
@@ -312,7 +314,8 @@ final class RexNodeJsonSerializer extends StdSerializer<RexNode> {
gen.writeEndObject();
}
gen.writeEndArray();
- gen.writeBooleanField(FIELD_NAME_CONTAINS_NULL, value.containsNull);
+ final SerializableSymbol symbol = calciteToSerializable(value.nullAs);
+ gen.writeStringField(FIELD_NAME_NULL_AS, symbol.getValue());
gen.writeEndObject();
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/SymbolUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/SymbolUtil.java
index 8724ea45cfa..1d9ab77b635 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/SymbolUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/SymbolUtil.java
@@ -33,6 +33,7 @@ import org.apache.flink.table.utils.DateTimeUtils;
import com.google.common.collect.BoundType;
import org.apache.calcite.avatica.util.TimeUnit;
import org.apache.calcite.avatica.util.TimeUnitRange;
+import org.apache.calcite.rex.RexUnknownAs;
import org.apache.calcite.sql.SqlJsonConstructorNullClause;
import org.apache.calcite.sql.SqlJsonEmptyOrError;
import org.apache.calcite.sql.SqlJsonExistsErrorBehavior;
@@ -473,6 +474,11 @@ public final class SymbolUtil {
// BOUND
addSymbolMapping(null, null, BoundType.OPEN, "BOUND", "OPEN");
addSymbolMapping(null, null, BoundType.CLOSED, "BOUND", "CLOSED");
+
+ // UNKNOWN_AS
+ addSymbolMapping(null, null, RexUnknownAs.TRUE, "UNKNOWN_AS", "TRUE");
+ addSymbolMapping(null, null, RexUnknownAs.FALSE, "UNKNOWN_AS", "FALSE");
+ addSymbolMapping(null, null, RexUnknownAs.UNKNOWN, "UNKNOWN_AS", "UNKNOWN");
}
/**
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 eddbcab7247..bdeec088414 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
@@ -9,8 +9,8 @@ This project bundles the following dependencies under the Apache Software Licens
- com.google.guava:guava:29.0-jre
- com.google.guava:failureaccess:1.0.1
- com.esri.geometry:esri-geometry-api:2.2.0
-- org.apache.calcite:calcite-core:1.29.0
-- org.apache.calcite:calcite-linq4j:1.29.0
+- org.apache.calcite:calcite-core:1.30.0
+- org.apache.calcite:calcite-linq4j:1.30.0
- org.apache.calcite.avatica:avatica-core:1.20.0
- commons-codec:commons-codec:1.15
- commons-io:commons-io:2.11.0
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkLogicalRelFactories.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkLogicalRelFactories.scala
index a7e7ff3ea2a..652a177bb98 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkLogicalRelFactories.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkLogicalRelFactories.scala
@@ -174,6 +174,7 @@ object FlinkLogicalRelFactories {
def createCorrelate(
left: RelNode,
right: RelNode,
+ hints: util.List[RelHint],
correlationId: CorrelationId,
requiredColumns: ImmutableBitSet,
joinType: JoinRelType): RelNode = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala
index e7cfb468a60..82fd2a9a682 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala
@@ -422,7 +422,11 @@ class FlinkTypeFactory(
new GenericRelDataType(generic.genericType, isNullable, typeSystem)
case it: TimeIndicatorRelDataType =>
- new TimeIndicatorRelDataType(it.typeSystem, it.originalType, isNullable, it.isEventTime)
+ new TimeIndicatorRelDataType(
+ it.typeSystemField,
+ it.originalType,
+ isNullable,
+ it.isEventTime)
// for nested rows we keep the nullability property,
// top-level rows fall back to Calcite's default handling
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/PreValidateReWriter.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/PreValidateReWriter.scala
index 47dd5734402..1ba6643dc29 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/PreValidateReWriter.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/PreValidateReWriter.scala
@@ -23,6 +23,7 @@ import org.apache.flink.sql.parser.dml.RichSqlInsert
import org.apache.flink.sql.parser.dql.SqlRichExplain
import org.apache.flink.table.api.ValidationException
import org.apache.flink.table.planner.calcite.PreValidateReWriter.{appendPartitionAndNullsProjects, notSupported}
+import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable
import org.apache.flink.table.planner.plan.schema.{CatalogSourceTable, FlinkPreparingTableBase, LegacyCatalogSourceTable}
import org.apache.flink.util.Preconditions.checkArgument
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/TimeIndicatorRelDataType.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/TimeIndicatorRelDataType.scala
index 3789f24ab9b..a4076db6adf 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/TimeIndicatorRelDataType.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/schema/TimeIndicatorRelDataType.scala
@@ -28,11 +28,11 @@ import java.lang
* basic SQL type.
*/
class TimeIndicatorRelDataType(
- val typeSystem: RelDataTypeSystem,
+ val typeSystemField: RelDataTypeSystem,
val originalType: BasicSqlType,
val nullable: Boolean,
val isEventTime: Boolean)
- extends BasicSqlType(typeSystem, originalType.getSqlTypeName, originalType.getPrecision) {
+ extends BasicSqlType(typeSystemField, originalType.getSqlTypeName, originalType.getPrecision) {
this.isNullable = nullable
computeDigest()
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest.java
index 23652b1fcad..b763d92cdec 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest.java
@@ -114,4 +114,18 @@ public class CalcJsonPlanTest extends TableTestBase {
+ "from MyTable where "
+ "(udf1(a) > 0 or (a * b) < 100) and b > 10");
}
+
+ @Test
+ public void testSarg() {
+ String sinkTableDdl =
+ "CREATE TABLE MySink (\n"
+ + " a bigint\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'sink-insert-only' = 'false',\n"
+ + " 'table-sink-class' = 'DEFAULT')";
+ tEnv.executeSql(sinkTableDdl);
+ String sql = "insert into MySink SELECT a from MyTable where a = 1 or a = 2 or a is null";
+ util.verifyJsonPlan(sql);
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/SargJsonPlanITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/SargJsonPlanITCase.java
new file mode 100644
index 00000000000..1320fa24920
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/jsonplan/SargJsonPlanITCase.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.runtime.stream.jsonplan;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.utils.JsonPlanTestBase;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+/** Test for Sarg JsonPlan ser/de. */
+public class SargJsonPlanITCase extends JsonPlanTestBase {
+ @Test
+ public void testSarg() throws ExecutionException, InterruptedException {
+ List<Row> data =
+ Arrays.asList(Row.of(1), Row.of(2), Row.of((Integer) null), Row.of(4), Row.of(5));
+ createTestValuesSourceTable("MyTable", data, "a int");
+ createTestNonInsertOnlyValuesSinkTable("`result`", "a int");
+ String sql =
+ "insert into `result` SELECT a\n"
+ + "FROM MyTable WHERE a = 1 OR a = 2 OR a IS NULL";
+ compileSqlAndExecutePlan(sql).await();
+ List<String> expected = Arrays.asList("+I[1]", "+I[2]", "+I[null]");
+ assertResult(expected, TestValuesTableFactory.getResults("result"));
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest_jsonplan/testSarg.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest_jsonplan/testSarg.out
new file mode 100644
index 00000000000..f39a0d6b95d
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcJsonPlanTest_jsonplan/testSarg.out
@@ -0,0 +1,157 @@
+{
+ "flinkVersion" : "",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MyTable`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "d",
+ "dataType" : "TIMESTAMP(3)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ],
+ "options" : {
+ "connector" : "values",
+ "bounded" : "false"
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "FilterPushDown",
+ "predicates" : [ ]
+ }, {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ] ],
+ "producedType" : "ROW<`a` BIGINT> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`a` BIGINT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`a` BIGINT>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, filter=[], project=[a], metadata=[]]], fields=[a])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ } ],
+ "condition" : {
+ "kind" : "CALL",
+ "syntax" : "INTERNAL",
+ "internalName" : "$SEARCH$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "LITERAL",
+ "sarg" : {
+ "ranges" : [ {
+ "lower" : {
+ "value" : 1,
+ "boundType" : "CLOSED"
+ },
+ "upper" : {
+ "value" : 1,
+ "boundType" : "CLOSED"
+ }
+ }, {
+ "lower" : {
+ "value" : 2,
+ "boundType" : "CLOSED"
+ },
+ "upper" : {
+ "value" : 2,
+ "boundType" : "CLOSED"
+ }
+ } ],
+ "nullAs" : "TRUE"
+ },
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` BIGINT>",
+ "description" : "Calc(select=[a], where=[SEARCH(a, Sarg[1, 2; NULL AS TRUE])])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "BIGINT"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ],
+ "options" : {
+ "sink-insert-only" : "false",
+ "table-sink-class" : "DEFAULT",
+ "connector" : "values"
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
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 06e50d09bb2..c3ed918655d 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
@@ -6,6 +6,6 @@ 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.6.0
+- com.jayway.jsonpath:json-path:2.7.0
- org.codehaus.janino:janino:3.1.9
- org.codehaus.janino:commons-compiler:3.1.9
diff --git a/flink-table/pom.xml b/flink-table/pom.xml
index ab5e343fdfb..0efbed1fad7 100644
--- a/flink-table/pom.xml
+++ b/flink-table/pom.xml
@@ -77,12 +77,12 @@ under the License.
</dependencyManagement>
<properties>
- <calcite.version>1.29.0</calcite.version>
- <!-- Calcite 1.29.0 depends on 3.1.6,
+ <calcite.version>1.30.0</calcite.version>
+ <!-- Calcite 1.30.0 depends on 3.1.6,
at the same time minimum 3.1.x Janino version passing Flink tests is 3.1.9,
more details are in FLINK-27995 -->
<janino.version>3.1.9</janino.version>
- <jsonpath.version>2.6.0</jsonpath.version>
+ <jsonpath.version>2.7.0</jsonpath.version>
<guava.version>29.0-jre</guava.version>
</properties>
</project>
[flink] 02/03: [hotfix] replace deprecated calcite apis
Posted by sn...@apache.org.
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 57f8d58614f5c5355d1a4780fc7c08b80a488061
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Thu Dec 8 18:52:56 2022 +0100
[hotfix] replace deprecated calcite apis
---
.../planner/delegation/hive/HiveParserCalcitePlanner.java | 11 +++++++++--
.../planner/delegation/hive/HiveParserRexNodeConverter.java | 3 +--
.../delegation/hive/copy/HiveParserSqlCountAggFunction.java | 7 ++++++-
.../delegation/hive/copy/HiveParserSqlMinMaxAggFunction.java | 7 ++++++-
.../delegation/hive/copy/HiveParserSqlSumAggFunction.java | 7 ++++++-
.../table/planner/expressions/converter/OverConvertRule.java | 3 ++-
.../flink/table/planner/functions/sql/SqlListAggFunction.java | 4 +++-
.../functions/sql/internal/SqlAuxiliaryGroupAggFunction.java | 4 +++-
.../logical/FlinkAggregateExpandDistinctAggregatesRule.java | 2 +-
.../rules/logical/FlinkSemiAntiJoinFilterTransposeRule.java | 4 +++-
.../flink/table/planner/plan/metadata/FlinkRelMdSize.scala | 6 ++++--
.../planner/plan/rules/logical/CalcRankTransposeRule.scala | 2 +-
.../plan/rules/logical/DecomposeGroupingSetsRule.scala | 2 +-
.../planner/plan/rules/logical/FlinkRewriteSubQueryRule.scala | 2 +-
.../planner/plan/rules/logical/FlinkSubQueryRemoveRule.scala | 4 ++--
.../plan/rules/logical/ProjectSemiAntiJoinTransposeRule.scala | 2 +-
.../table/planner/plan/rules/logical/SplitAggregateRule.scala | 4 +---
.../flink/table/planner/plan/trait/FlinkRelDistribution.scala | 2 +-
.../org/apache/flink/table/planner/plan/trait/TraitUtil.scala | 2 +-
.../plan/metadata/AggCallSelectivityEstimatorTest.scala | 2 ++
.../flink/table/planner/plan/utils/FlinkRexUtilTest.scala | 2 +-
21 files changed, 56 insertions(+), 26 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserCalcitePlanner.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserCalcitePlanner.java
index 1a362b82f7a..63546109963 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserCalcitePlanner.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserCalcitePlanner.java
@@ -60,6 +60,8 @@ import org.apache.flink.table.planner.plan.nodes.hive.LogicalDistribution;
import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
+
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.plan.ViewExpanders;
@@ -1218,7 +1220,8 @@ public class HiveParserCalcitePlanner {
gbInputRexNodes.add(cluster.getRexBuilder().makeInputRef(srcRel, 0));
}
- return LogicalAggregate.create(gbInputRel, groupSet, transformedGroupSets, aggregateCalls);
+ return LogicalAggregate.create(
+ gbInputRel, ImmutableList.of(), groupSet, transformedGroupSets, aggregateCalls);
}
// Generate GB plan.
@@ -2409,7 +2412,11 @@ public class HiveParserCalcitePlanner {
ImmutableBitSet.range(res.getRowType().getFieldList().size());
res =
LogicalAggregate.create(
- res, groupSet, Collections.emptyList(), Collections.emptyList());
+ res,
+ ImmutableList.of(),
+ groupSet,
+ Collections.emptyList(),
+ Collections.emptyList());
HiveParserRowResolver groupByOutputRowResolver = new HiveParserRowResolver();
for (int i = 0; i < outRR.getColumnInfos().size(); i++) {
ColumnInfo colInfo = outRR.getColumnInfos().get(i);
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserRexNodeConverter.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserRexNodeConverter.java
index a12570268a2..5c5f67aa68e 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserRexNodeConverter.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserRexNodeConverter.java
@@ -761,8 +761,7 @@ public class HiveParserRexNodeConverter {
.makeNullLiteral(
newChildRexNodeLst
.get(newChildRexNodeLst.size() - 1)
- .getType()
- .getSqlTypeName()));
+ .getType()));
}
return newChildRexNodeLst;
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlCountAggFunction.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlCountAggFunction.java
index 40419c04e88..fdf25dcf8ff 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlCountAggFunction.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlCountAggFunction.java
@@ -30,6 +30,7 @@ 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.util.ImmutableIntList;
+import org.apache.calcite.util.Optionality;
/**
* Counterpart of hive's
@@ -50,11 +51,15 @@ public class HiveParserSqlCountAggFunction extends SqlAggFunction
SqlOperandTypeChecker operandTypeChecker) {
super(
"count",
+ null,
SqlKind.COUNT,
returnTypeInference,
operandTypeInference,
operandTypeChecker,
- SqlFunctionCategory.NUMERIC);
+ SqlFunctionCategory.NUMERIC,
+ false,
+ false,
+ Optionality.FORBIDDEN);
this.isDistinct = isDistinct;
this.returnTypeInference = returnTypeInference;
this.operandTypeChecker = operandTypeChecker;
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlMinMaxAggFunction.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlMinMaxAggFunction.java
index 996407be806..3c7de168fe9 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlMinMaxAggFunction.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlMinMaxAggFunction.java
@@ -25,6 +25,7 @@ import org.apache.calcite.sql.SqlSplittableAggFunction;
import org.apache.calcite.sql.type.SqlOperandTypeChecker;
import org.apache.calcite.sql.type.SqlOperandTypeInference;
import org.apache.calcite.sql.type.SqlReturnTypeInference;
+import org.apache.calcite.util.Optionality;
/**
* Counterpart of hive's
@@ -39,11 +40,15 @@ public class HiveParserSqlMinMaxAggFunction extends SqlAggFunction {
boolean isMin) {
super(
isMin ? "min" : "max",
+ null,
isMin ? SqlKind.MIN : SqlKind.MAX,
returnTypeInference,
operandTypeInference,
operandTypeChecker,
- SqlFunctionCategory.NUMERIC);
+ SqlFunctionCategory.NUMERIC,
+ false,
+ false,
+ Optionality.FORBIDDEN);
}
@Override
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlSumAggFunction.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlSumAggFunction.java
index 11362dbaa34..6f860b0e46e 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlSumAggFunction.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSqlSumAggFunction.java
@@ -36,6 +36,7 @@ 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.util.ImmutableIntList;
+import org.apache.calcite.util.Optionality;
import java.util.ArrayList;
import java.util.Collections;
@@ -62,11 +63,15 @@ public class HiveParserSqlSumAggFunction extends SqlAggFunction
SqlOperandTypeChecker operandTypeChecker) {
super(
"sum",
+ null,
SqlKind.SUM,
returnTypeInference,
operandTypeInference,
operandTypeChecker,
- SqlFunctionCategory.NUMERIC);
+ SqlFunctionCategory.NUMERIC,
+ false,
+ false,
+ Optionality.FORBIDDEN);
this.returnTypeInference = returnTypeInference;
this.operandTypeChecker = operandTypeChecker;
this.operandTypeInference = operandTypeInference;
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
index c5fff324c9a..72a50297bcb 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/OverConvertRule.java
@@ -133,7 +133,8 @@ public class OverConvertRule implements CallExpressionConvertRule {
isPhysical,
true,
false,
- isDistinct));
+ isDistinct,
+ false));
}
return Optional.empty();
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlListAggFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlListAggFunction.java
index 15511623585..c04fd4ca51e 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlListAggFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/SqlListAggFunction.java
@@ -29,6 +29,7 @@ import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.type.OperandTypes;
import org.apache.calcite.sql.type.ReturnTypes;
import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.Optionality;
import java.util.List;
@@ -57,7 +58,8 @@ public class SqlListAggFunction extends SqlAggFunction {
OperandTypes.and(OperandTypes.CHARACTER, OperandTypes.LITERAL))),
SqlFunctionCategory.SYSTEM,
false,
- false);
+ false,
+ Optionality.FORBIDDEN);
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/internal/SqlAuxiliaryGroupAggFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/internal/SqlAuxiliaryGroupAggFunction.java
index 11dcc1072e6..b2a4ef47d70 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/internal/SqlAuxiliaryGroupAggFunction.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/internal/SqlAuxiliaryGroupAggFunction.java
@@ -25,6 +25,7 @@ import org.apache.calcite.sql.SqlFunctionCategory;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.type.OperandTypes;
import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.util.Optionality;
/**
* An internal [[SqlAggFunction]] to represents auxiliary group keys which will not be computed as
@@ -43,6 +44,7 @@ public class SqlAuxiliaryGroupAggFunction extends SqlAggFunction {
OperandTypes.ANY,
SqlFunctionCategory.SYSTEM,
false,
- false);
+ false,
+ Optionality.FORBIDDEN);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateExpandDistinctAggregatesRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateExpandDistinctAggregatesRule.java
index 57f941ea410..2f885852598 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateExpandDistinctAggregatesRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateExpandDistinctAggregatesRule.java
@@ -470,7 +470,7 @@ public final class FlinkAggregateExpandDistinctAggregatesRule extends RelOptRule
aggCall.left.filterArg,
aggregate.getGroupCount(),
fullGroupSet.cardinality());
- distinctAggCalls.add(newAggCall.rename(aggCall.right));
+ distinctAggCalls.add(newAggCall.withName(aggCall.right));
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkSemiAntiJoinFilterTransposeRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkSemiAntiJoinFilterTransposeRule.java
index c77097308c5..a3d6de0f7ec 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkSemiAntiJoinFilterTransposeRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkSemiAntiJoinFilterTransposeRule.java
@@ -17,6 +17,8 @@
package org.apache.flink.table.planner.plan.rules.logical;
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
+
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.rel.RelNode;
@@ -79,7 +81,7 @@ public class FlinkSemiAntiJoinFilterTransposeRule extends RelOptRule {
join.getJoinType());
final RelFactories.FilterFactory factory = RelFactories.DEFAULT_FILTER_FACTORY;
- RelNode newFilter = factory.createFilter(newJoin, filter.getCondition());
+ RelNode newFilter = factory.createFilter(newJoin, filter.getCondition(), ImmutableSet.of());
call.transformTo(newFilter);
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
index 0f312c35bfa..9a16bb3756d 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdSize.scala
@@ -123,10 +123,12 @@ class FlinkRelMdSize private extends MetadataHandler[BuiltInMetadata.Size] {
def averageColumnSizes(rel: Expand, mq: RelMetadataQuery): JList[JDouble] = {
val fieldCount = rel.getRowType.getFieldCount
+ val fieldList = rel.getRowType.getFieldList
// get each column's RexNode (RexLiteral, RexInputRef or null)
val projectNodes = (0 until fieldCount).map {
i =>
- val initNode: RexNode = rel.getCluster.getRexBuilder.constantNull()
+ val initNode: RexNode =
+ rel.getCluster.getRexBuilder.makeNullLiteral(fieldList.get(i).getType)
rel.projects.foldLeft(initNode) {
(mergeNode, project) =>
(mergeNode, project.get(i)) match {
@@ -149,7 +151,7 @@ class FlinkRelMdSize private extends MetadataHandler[BuiltInMetadata.Size] {
case (p, i) =>
val size = if (p == null || i == rel.expandIdIndex) {
// use default value
- FlinkRelMdSize.averageTypeValueSize(rel.getRowType.getFieldList.get(i).getType)
+ FlinkRelMdSize.averageTypeValueSize(fieldList.get(i).getType)
} else {
// use value from input
averageRexSize(p, inputColumnSizes)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CalcRankTransposeRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CalcRankTransposeRule.scala
index 454d80fd951..74b611e6330 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CalcRankTransposeRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CalcRankTransposeRule.scala
@@ -166,7 +166,7 @@ class CalcRankTransposeRule
val oldOrderKey = rank.orderKey
val oldFieldCollations = oldOrderKey.getFieldCollations
val newFieldCollations = oldFieldCollations.map {
- fc => fc.copy(fieldMapping(fc.getFieldIndex))
+ fc => fc.withFieldIndex(fieldMapping(fc.getFieldIndex))
}
val newOrderKey = if (newFieldCollations.eq(oldFieldCollations)) {
oldOrderKey
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/DecomposeGroupingSetsRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/DecomposeGroupingSetsRule.scala
index 723888c436d..140c1d7c92c 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/DecomposeGroupingSetsRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/DecomposeGroupingSetsRule.scala
@@ -328,7 +328,7 @@ class DecomposeGroupingSetsRule
val res: Long = call.getArgList.foldLeft(0L)(
(res, arg) => (res << 1L) + (if (groups.contains(arg)) 0L else 1L))
builder.makeLiteral(res, call.getType, false)
- case _ => builder.constantNull()
+ case _ => builder.makeNullLiteral(call.getType)
}
}
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkRewriteSubQueryRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkRewriteSubQueryRule.scala
index e198a1f44c0..3cee597cd1a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkRewriteSubQueryRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkRewriteSubQueryRule.scala
@@ -56,7 +56,7 @@ class FlinkRewriteSubQueryRule(
val filter: Filter = call.rel(0)
val condition = filter.getCondition
val newCondition = rewriteScalarQuery(condition)
- if (RexUtil.eq(condition, newCondition)) {
+ if (condition.equals(newCondition)) {
return
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkSubQueryRemoveRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkSubQueryRemoveRule.scala
index 62607017313..518afbf7df7 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkSubQueryRemoveRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkSubQueryRemoveRule.scala
@@ -304,13 +304,13 @@ class FlinkSubQueryRemoveRule(
replacement: RexNode): RexNode = {
condition.accept(new RexShuttle() {
override def visitSubQuery(subQuery: RexSubQuery): RexNode = {
- if (RexUtil.eq(subQuery, oldSubQueryCall)) replacement else subQuery
+ if (subQuery.equals(oldSubQueryCall)) replacement else subQuery
}
override def visitCall(call: RexCall): RexNode = {
call.getKind match {
case SqlKind.NOT if call.operands.head.isInstanceOf[RexSubQuery] =>
- if (RexUtil.eq(call, oldSubQueryCall)) replacement else call
+ if (call.equals(oldSubQueryCall)) replacement else call
case _ =>
super.visitCall(call)
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRule.scala
index 96b86be053a..f912b020919 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRule.scala
@@ -108,7 +108,7 @@ class ProjectSemiAntiJoinTransposeRule
inputNeededFields: ImmutableBitSet,
offset: Int): RelNode = {
val rexBuilder = originInput.getCluster.getRexBuilder
- val typeBuilder = new RelDataTypeFactory.FieldInfoBuilder(relBuilder.getTypeFactory)
+ val typeBuilder = relBuilder.getTypeFactory.builder()
val newProjects: util.List[RexNode] = new util.ArrayList[RexNode]()
val newFieldNames: util.List[String] = new util.ArrayList[String]()
inputNeededFields.toList.foreach {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/SplitAggregateRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/SplitAggregateRule.scala
index 35f8579b609..5ca5526059d 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/SplitAggregateRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/SplitAggregateRule.scala
@@ -384,9 +384,7 @@ class SplitAggregateRule
FlinkSqlOperatorTable.EQUALS,
countInputRef,
relBuilder.getRexBuilder.makeBigintLiteral(JBigDecimal.valueOf(0)))
- val ifTrue = relBuilder.cast(
- relBuilder.getRexBuilder.constantNull(),
- aggCall.`type`.getSqlTypeName)
+ val ifTrue = relBuilder.getRexBuilder.makeNullLiteral(aggCall.`type`)
val ifFalse = relBuilder.call(FlinkSqlOperatorTable.DIVIDE, sumInputRef, countInputRef)
relBuilder.call(FlinkSqlOperatorTable.IF, equals, ifTrue, ifFalse)
} else {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/FlinkRelDistribution.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/FlinkRelDistribution.scala
index e7fef969b13..1e1e58953c7 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/FlinkRelDistribution.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/FlinkRelDistribution.scala
@@ -119,7 +119,7 @@ class FlinkRelDistribution private (
try {
val i = mapping.getTargetOpt(fieldCollation.getFieldIndex)
if (i >= 0) {
- newFieldCollations.add(fieldCollation.copy(i))
+ newFieldCollations.add(fieldCollation.withFieldIndex(i))
} else {
return FlinkRelDistribution.ANY
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/TraitUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/TraitUtil.scala
index 247514c6e68..ba42dff0bef 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/TraitUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/TraitUtil.scala
@@ -49,7 +49,7 @@ object TraitUtil {
fieldCollation =>
try {
val i = mapping.getTargetOpt(fieldCollation.getFieldIndex)
- if (i >= 0) newFieldCollations.add(fieldCollation.copy(i))
+ if (i >= 0) newFieldCollations.add(fieldCollation.withFieldIndex(i))
else return RelCollations.EMPTY
} catch {
case _: IndexOutOfBoundsException => return RelCollations.EMPTY
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala
index 4d912ebc6df..ae27e1d7f2b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala
@@ -30,6 +30,7 @@ import org.apache.calcite.jdbc.CalciteSchema
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.RelCollations
import org.apache.calcite.rel.core.{Aggregate, AggregateCall, TableScan}
+import org.apache.calcite.rel.hint.RelHint
import org.apache.calcite.rel.logical.LogicalAggregate
import org.apache.calcite.rel.metadata.{JaninoRelMetadataProvider, RelMetadataQueryBase}
import org.apache.calcite.rex.{RexInputRef, RexLiteral, RexNode}
@@ -118,6 +119,7 @@ class AggCallSelectivityEstimatorTest {
LogicalAggregate.create(
scan,
+ ImmutableList.of[RelHint],
ImmutableBitSet.of(groupSet: _*),
null,
ImmutableList.copyOf(aggCalls.toArray))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala
index 5e7ba810fed..28744d87c2e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala
@@ -206,7 +206,7 @@ class FlinkRexUtilTest {
assertEquals(RexUtil.toCnf(rexBuilder, predicate).toString, newPredicate3.toString)
val newPredicate4 = FlinkRexUtil.toCnf(rexBuilder, Int.MaxValue, predicate)
- assertFalse(RexUtil.eq(predicate, newPredicate4))
+ assertFalse(predicate.equals(newPredicate4))
assertEquals(RexUtil.toCnf(rexBuilder, predicate).toString, newPredicate4.toString)
}
[flink] 01/03: [FLINK-29214][Table API/SQL] Remove usages of deprecated Aggregate#indicator
Posted by sn...@apache.org.
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 0412329dea41d8f1325b6eefc2ab0914142be007
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Tue Sep 6 23:29:16 2022 +0200
[FLINK-29214][Table API/SQL] Remove usages of deprecated Aggregate#indicator
---
.../planner/plan/rules/logical/FlinkAggregateJoinTransposeRule.java | 1 -
.../table/planner/plan/rules/logical/FlinkAggregateRemoveRule.java | 4 +---
.../rules/physical/stream/StreamPhysicalPythonGroupAggregateRule.java | 2 +-
.../physical/stream/StreamPhysicalPythonGroupWindowAggregateRule.java | 2 +-
.../planner/plan/rules/logical/AggregateReduceGroupingRule.scala | 3 +--
.../planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala | 3 ++-
.../table/planner/plan/rules/logical/PruneAggregateCallRule.scala | 3 +--
.../planner/plan/rules/physical/batch/BatchPhysicalHashAggRule.scala | 4 ----
.../plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala | 2 +-
.../plan/rules/physical/stream/StreamPhysicalGroupAggregateRule.scala | 2 +-
.../physical/stream/StreamPhysicalGroupWindowAggregateRule.scala | 3 +--
.../rules/physical/stream/StreamPhysicalWindowAggregateRule.scala | 2 +-
12 files changed, 11 insertions(+), 20 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateJoinTransposeRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateJoinTransposeRule.java
index 72e2b3e94e9..3f9415ea0c0 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateJoinTransposeRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateJoinTransposeRule.java
@@ -488,7 +488,6 @@ public class FlinkAggregateJoinTransposeRule extends RelOptRule {
aggregate.copy(
aggregate.getTraitSet(),
aggregate.getInput(),
- aggregate.indicator,
newGroupSet,
com.google.common.collect.ImmutableList.of(newGroupSet),
aggCalls);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateRemoveRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateRemoveRule.java
index 820156c8261..8f9227cd4b6 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateRemoveRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkAggregateRemoveRule.java
@@ -73,9 +73,7 @@ public class FlinkAggregateRemoveRule extends RelOptRule {
public boolean matches(RelOptRuleCall call) {
final Aggregate aggregate = call.rel(0);
final RelNode input = call.rel(1);
- if (aggregate.getGroupCount() == 0
- || aggregate.indicator
- || aggregate.getGroupType() != Aggregate.Group.SIMPLE) {
+ if (aggregate.getGroupCount() == 0 || aggregate.getGroupType() != Aggregate.Group.SIMPLE) {
return false;
}
for (AggregateCall aggCall : aggregate.getAggCallList()) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupAggregateRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupAggregateRule.java
index acae431f2a7..0bebe9aca32 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupAggregateRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupAggregateRule.java
@@ -62,7 +62,7 @@ public class StreamPhysicalPythonGroupAggregateRule extends ConverterRule {
FlinkLogicalAggregate agg = call.rel(0);
// check if we have grouping sets
- if (agg.getGroupType() != Aggregate.Group.SIMPLE || agg.indicator) {
+ if (agg.getGroupType() != Aggregate.Group.SIMPLE) {
throw new TableException("GROUPING SETS are currently not supported.");
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupWindowAggregateRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupWindowAggregateRule.java
index cdb2f54d3cb..8d2d662ea31 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupWindowAggregateRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalPythonGroupWindowAggregateRule.java
@@ -69,7 +69,7 @@ public class StreamPhysicalPythonGroupWindowAggregateRule extends ConverterRule
List<AggregateCall> aggCalls = agg.getAggCallList();
// check if we have grouping sets
- if (agg.getGroupType() != Aggregate.Group.SIMPLE || agg.indicator) {
+ if (agg.getGroupType() != Aggregate.Group.SIMPLE) {
throw new TableException("GROUPING SETS are currently not supported.");
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/AggregateReduceGroupingRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/AggregateReduceGroupingRule.scala
index d1845c685af..6ee02c5e1da 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/AggregateReduceGroupingRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/AggregateReduceGroupingRule.scala
@@ -44,7 +44,7 @@ class AggregateReduceGroupingRule(relBuilderFactory: RelBuilderFactory)
override def matches(call: RelOptRuleCall): Boolean = {
val agg: Aggregate = call.rel(0)
- agg.getGroupCount > 1 && agg.getGroupType == Group.SIMPLE && !agg.indicator
+ agg.getGroupCount > 1 && agg.getGroupType == Group.SIMPLE
}
override def onMatch(call: RelOptRuleCall): Unit = {
@@ -106,7 +106,6 @@ class AggregateReduceGroupingRule(relBuilderFactory: RelBuilderFactory)
val newAgg = agg.copy(
agg.getTraitSet,
input,
- agg.indicator, // always false here
newGrouping,
ImmutableList.of(newGrouping),
newAggCalls
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
index 901c136cfe1..33a1862142f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalWindowAggregateRuleBase.scala
@@ -36,6 +36,7 @@ import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.core.{Aggregate, AggregateCall, RelFactories}
import org.apache.calcite.rel.core.Aggregate.Group
+import org.apache.calcite.rel.hint.RelHint
import org.apache.calcite.rel.logical.{LogicalAggregate, LogicalProject}
import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.SqlTypeUtil
@@ -98,7 +99,7 @@ abstract class LogicalWindowAggregateRuleBase(description: String)
// we don't use the builder here because it uses RelMetadataQuery which affects the plan
val newAgg = LogicalAggregate.create(
newProject,
- agg.indicator,
+ ImmutableList.of[RelHint](),
newGroupSet,
ImmutableList.of(newGroupSet),
finalCalls)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PruneAggregateCallRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PruneAggregateCallRule.scala
index ee378cc4ffb..bd7c479fea3 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PruneAggregateCallRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PruneAggregateCallRule.scala
@@ -45,7 +45,7 @@ abstract class PruneAggregateCallRule[T <: RelNode](topClass: Class[T])
val relOnAgg: T = call.rel(0)
val agg: Aggregate = call.rel(1)
if (
- agg.indicator || agg.getGroupType != Group.SIMPLE || agg.getAggCallList.isEmpty ||
+ agg.getGroupType != Group.SIMPLE || agg.getAggCallList.isEmpty ||
// at least output one column
(agg.getGroupCount == 0 && agg.getAggCallList.size() == 1)
) {
@@ -91,7 +91,6 @@ abstract class PruneAggregateCallRule[T <: RelNode](topClass: Class[T])
val newAgg = agg.copy(
agg.getTraitSet,
agg.getInput,
- agg.indicator,
agg.getGroupSet,
ImmutableList.of(agg.getGroupSet),
newAggCalls
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashAggRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashAggRule.scala
index f6ad3a06f89..355c1d0779a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashAggRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashAggRule.scala
@@ -79,10 +79,6 @@ class BatchPhysicalHashAggRule
val input: RelNode = call.rel(1)
val inputRowType = input.getRowType
- if (agg.indicator) {
- throw new UnsupportedOperationException("Not support group sets aggregate now.")
- }
-
val groupSet = agg.getGroupSet.toArray
val (auxGroupSet, aggCallsWithoutAuxGroupCalls) = AggregateUtil.checkAndSplitAggCalls(agg)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala
index 13ddd5c12c6..0402b0bd861 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalWindowAggregateRule.scala
@@ -84,7 +84,7 @@ class BatchPhysicalWindowAggregateRule
// check if we have grouping sets
val groupSets = agg.getGroupType != Group.SIMPLE
- if (groupSets || agg.indicator) {
+ if (groupSets) {
throw new TableException("GROUPING SETS are currently not supported.")
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupAggregateRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupAggregateRule.scala
index a7f71c2e515..13ad78c5850 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupAggregateRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupAggregateRule.scala
@@ -41,7 +41,7 @@ class StreamPhysicalGroupAggregateRule(config: Config) extends ConverterRule(con
val agg: FlinkLogicalAggregate = call.rel(0)
// check if we have grouping sets
- if (agg.getGroupType != Group.SIMPLE || agg.indicator) {
+ if (agg.getGroupType != Group.SIMPLE) {
throw new TableException("GROUPING SETS are currently not supported.")
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupWindowAggregateRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupWindowAggregateRule.scala
index 9bfaebdd07c..13998116eac 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalGroupWindowAggregateRule.scala
@@ -43,8 +43,7 @@ class StreamPhysicalGroupWindowAggregateRule(config: Config) extends ConverterRu
val agg: FlinkLogicalWindowAggregate = call.rel(0)
// check if we have grouping sets
- val groupSets = agg.getGroupType != Group.SIMPLE
- if (groupSets || agg.indicator) {
+ if (agg.getGroupType != Group.SIMPLE) {
throw new TableException("GROUPING SETS are currently not supported.")
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalWindowAggregateRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalWindowAggregateRule.scala
index 4778c4c9c0a..aec1b4dd32d 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalWindowAggregateRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalWindowAggregateRule.scala
@@ -46,7 +46,7 @@ class StreamPhysicalWindowAggregateRule(config: Config) extends ConverterRule(co
val agg: FlinkLogicalAggregate = call.rel(0)
// check if we have grouping sets
- if (agg.getGroupType != Group.SIMPLE || agg.indicator) {
+ if (agg.getGroupType != Group.SIMPLE) {
throw new TableException("GROUPING SETS are currently not supported.")
}