You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2023/06/29 16:26:51 UTC
[flink] 01/03: [FLINK-28744][table] Upgrade Calcite to 1.31.0
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ab6a81118b45c06e822f3c77468dd7d0afb66b9e
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Mon May 8 01:11:29 2023 +0200
[FLINK-28744][table] Upgrade Calcite to 1.31.0
This updates the Flink code base to 1.31.0. See the pull request for a detailed
list of Calcite issues that caused plan changes.
Note that CALCITE-4861 has an impact on casting behavior. Some corner cases might
behave differently now. For example, casting from FLOAT/DOUBLE 9234567891.12 to
INT/BIGINT has now Java behavior for overflows.
This closes #22558.
---
flink-table/flink-sql-jdbc-driver/pom.xml | 6 +
flink-table/flink-sql-parser/pom.xml | 29 +-
.../src/main/codegen/includes/parserImpls.ftl | 31 +-
.../sql/parser/validate/FlinkSqlConformance.java | 5 +
.../flink/sql/parser/FlinkDDLDataTypeTest.java | 4 +-
flink-table/flink-table-calcite-bridge/pom.xml | 44 +-
flink-table/flink-table-planner/pom.xml | 8 +-
.../java/org/apache/calcite/rel/core/Filter.java | 212 ---------
.../org/apache/calcite/rel/core/Intersect.java | 77 ----
.../java/org/apache/calcite/rel/core/Minus.java | 69 ---
.../java/org/apache/calcite/rel/core/SetOp.java | 159 -------
.../java/org/apache/calcite/rel/core/Snapshot.java | 1 +
.../java/org/apache/calcite/rel/core/Sort.java | 291 -------------
.../java/org/apache/calcite/rel/core/Union.java | 79 ----
.../java/org/apache/calcite/rel/core/Values.java | 248 -----------
.../java/org/apache/calcite/rel/core/Window.java | 484 ---------------------
.../apache/calcite/rel/hint/HintPredicates.java | 140 ------
.../apache/calcite/rel/logical/LogicalFilter.java | 165 -------
.../calcite/rel/logical/LogicalIntersect.java | 97 -----
.../apache/calcite/rel/logical/LogicalMinus.java | 101 -----
.../apache/calcite/rel/logical/LogicalSort.java | 113 -----
.../apache/calcite/rel/logical/LogicalUnion.java | 101 -----
.../apache/calcite/rel/logical/LogicalValues.java | 160 -------
.../apache/calcite/rel/logical/LogicalWindow.java | 404 -----------------
.../calcite/sql/validate/SqlValidatorImpl.java | 302 ++++++++-----
.../apache/calcite/sql2rel/RelDecorrelator.java | 38 +-
.../apache/calcite/sql2rel/SqlToRelConverter.java | 387 ++++++++++++++--
.../java/org/apache/calcite/tools/RelBuilder.java | 66 ++-
.../src/main/resources/META-INF/NOTICE | 10 +-
.../table/planner/calcite/FlinkPlannerImpl.scala | 5 +-
.../planner/functions/CastFunctionITCase.java | 40 +-
.../plan/hints/batch/BroadcastJoinHintTest.xml | 8 +-
.../plan/hints/batch/NestLoopJoinHintTest.xml | 8 +-
.../plan/hints/batch/ShuffleHashJoinHintTest.xml | 8 +-
.../plan/hints/batch/ShuffleMergeJoinHintTest.xml | 8 +-
.../PushFilterInCalcIntoTableSourceRuleTest.xml | 4 +-
...PushFilterIntoLegacyTableSourceScanRuleTest.xml | 4 +-
.../PushFilterIntoTableSourceScanRuleTest.xml | 4 +-
.../SubqueryCorrelateVariablesValidationTest.xml | 32 +-
.../table/planner/plan/stream/table/CalcTest.xml | 4 +-
.../SubqueryCorrelateVariablesValidationTest.scala | 2 +-
flink-table/pom.xml | 6 +-
42 files changed, 785 insertions(+), 3179 deletions(-)
diff --git a/flink-table/flink-sql-jdbc-driver/pom.xml b/flink-table/flink-sql-jdbc-driver/pom.xml
index 32ed400537f..1b82576fb54 100644
--- a/flink-table/flink-sql-jdbc-driver/pom.xml
+++ b/flink-table/flink-sql-jdbc-driver/pom.xml
@@ -91,6 +91,12 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${guava.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/flink-table/flink-sql-parser/pom.xml b/flink-table/flink-sql-parser/pom.xml
index 4da406fe319..40b4260a3ea 100644
--- a/flink-table/flink-sql-parser/pom.xml
+++ b/flink-table/flink-sql-parser/pom.xml
@@ -62,11 +62,12 @@ under the License.
<version>${calcite.version}</version>
<exclusions>
<!--
- "mvn dependency:tree" as of Calcite 1.30.0:
+ "mvn dependency:tree" as of Calcite 1.31.0:
- [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
+ [INFO] +- org.apache.calcite:calcite-core:jar:1.31.0:compile
+ [INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.22.0:compile
+ [INFO] | +- org.apiguardian:apiguardian-api:jar:1.1.2:compile
+ [INFO] | \- org.checkerframework:checker-qual:jar:3.12.0:compile
Dependencies that are not needed for how we use Calcite right now.
-->
@@ -162,11 +163,6 @@ under the License.
<groupId>com.google.uzaygezen</groupId>
<artifactId>uzaygezen-core</artifactId>
</exclusion>
- <!-- Excluding org.apiguardian:apiguardian-api v1.1.0 because org.junit.jupiter:junit-jupiter contains newer version v1.1.2 -->
- <exclusion>
- <groupId>org.apiguardian</groupId>
- <artifactId>apiguardian-api</artifactId>
- </exclusion>
<exclusion>
<groupId>com.google.errorprone</groupId>
<artifactId>error_prone_annotations</artifactId>
@@ -179,6 +175,14 @@ under the License.
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib-jdk8</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>org.apache.httpcomponents.client5</groupId>
+ <artifactId>httpclient5</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.httpcomponents.core5</groupId>
+ <artifactId>httpcore5</artifactId>
+ </exclusion>
</exclusions>
</dependency>
<dependency>
@@ -226,13 +230,6 @@ under the License.
<artifactId>calcite-linq4j</artifactId>
<version>${calcite.version}</version>
<scope>test</scope>
- <exclusions>
- <!-- Excluding org.apiguardian:apiguardian-api v1.1.0 because org.junit.jupiter:junit-jupiter contains newer version v1.1.2 -->
- <exclusion>
- <groupId>org.apiguardian</groupId>
- <artifactId>apiguardian-api</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
diff --git a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
index 7aeedaed09d..701f172247b 100644
--- a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
+++ b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
@@ -1416,12 +1416,14 @@ SqlNode RichSqlInsert() :
final SqlNodeList keywordList;
final List<SqlLiteral> extendedKeywords = new ArrayList<SqlLiteral>();
final SqlNodeList extendedKeywordList;
- SqlNode table;
- SqlNodeList extendList = null;
+ final SqlIdentifier tableName;
+ SqlNode tableRef;
+ final SqlNodeList extendList;
SqlNode source;
final SqlNodeList partitionList = new SqlNodeList(getPos());
SqlNodeList columnList = null;
final Span s;
+ final Pair<SqlNodeList, SqlNodeList> p;
}
{
(
@@ -1445,31 +1447,28 @@ SqlNode RichSqlInsert() :
keywordList = new SqlNodeList(keywords, s.addAll(keywords).pos());
extendedKeywordList = new SqlNodeList(extendedKeywords, s.addAll(extendedKeywords).pos());
}
- table = TableRefWithHintsOpt()
- [
- LOOKAHEAD(5)
- [ <EXTEND> ]
- extendList = ExtendList() {
- table = extend(table, extendList);
- }
- ]
+ tableName = CompoundTableIdentifier()
+ ( tableRef = TableHints(tableName) | { tableRef = tableName; } )
+ [ LOOKAHEAD(5) tableRef = ExtendTable(tableRef) ]
[
<PARTITION> PartitionSpecCommaList(partitionList)
]
- [
+ (
LOOKAHEAD(2)
- { final Pair<SqlNodeList, SqlNodeList> p; }
p = ParenthesizedCompoundIdentifierList() {
if (p.right.size() > 0) {
- table = extend(table, p.right);
+ tableRef = extend(tableRef, p.right);
}
if (p.left.size() > 0) {
columnList = p.left;
+ } else {
+ columnList = null;
}
}
- ]
+ | { columnList = null; }
+ )
source = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY) {
- return new RichSqlInsert(s.end(source), keywordList, extendedKeywordList, table, source,
+ return new RichSqlInsert(s.end(source), keywordList, extendedKeywordList, tableRef, source,
columnList, partitionList);
}
}
@@ -2443,7 +2442,7 @@ SqlNodeList SimpleIdentifierCommaListWithPosition() :
}
{
{ s = span(); }
- SimpleIdentifierCommaList(list) {
+ AddSimpleIdentifiers(list) {
return new SqlNodeList(list, s.end(this));
}
}
diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/validate/FlinkSqlConformance.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/validate/FlinkSqlConformance.java
index d29cf226ab3..a04ce07d1cd 100644
--- a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/validate/FlinkSqlConformance.java
+++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/validate/FlinkSqlConformance.java
@@ -132,6 +132,11 @@ public enum FlinkSqlConformance implements SqlConformance {
return false;
}
+ @Override
+ public boolean isOffsetLimitAllowed() {
+ return false;
+ }
+
@Override
public boolean allowGeometry() {
return false;
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 3df21df7abc..ed558537ead 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
@@ -594,8 +594,8 @@ class FlinkDDLDataTypeTest {
}
private static SqlOperatorTable createOperatorTable(SqlOperatorTable opTab0) {
- MockSqlOperatorTable opTab = new MockSqlOperatorTable(opTab0);
- MockSqlOperatorTable.addRamp(opTab);
+ MockSqlOperatorTable opTab = MockSqlOperatorTable.of(opTab0);
+ opTab.extend();
return opTab;
}
diff --git a/flink-table/flink-table-calcite-bridge/pom.xml b/flink-table/flink-table-calcite-bridge/pom.xml
index 0323856a5f0..afebff7b29a 100644
--- a/flink-table/flink-table-calcite-bridge/pom.xml
+++ b/flink-table/flink-table-calcite-bridge/pom.xml
@@ -45,20 +45,21 @@ under the License.
<version>${calcite.version}</version>
<exclusions>
<!--
- "mvn dependency:tree" as of Calcite 1.30.0:
- [INFO] +- org.apache.calcite:calcite-core:jar:1.30.0:compile
- [INFO] | +- org.apache.calcite:calcite-linq4j:jar:1.30.0:compile
+ "mvn dependency:tree" as of Calcite 1.31.0:
+ [INFO] +- org.apache.calcite:calcite-core:jar:1.31.0:compile
+ [INFO] | +- org.apache.calcite:calcite-linq4j:jar:1.31.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
+ [INFO] | +- org.apache.calcite.avatica:avatica-core:jar:1.22.0:compile
+ [INFO] | +- org.apiguardian:apiguardian-api:jar:1.1.2:compile
[INFO] | +- com.fasterxml.jackson.core:jackson-core:jar:2.13.4:compile
[INFO] | +- com.fasterxml.jackson.core:jackson-databind:jar:2.13.4.2:compile
- [INFO] | +- com.jayway.jsonpath:json-path:jar:2.4.0:runtime
- [INFO] | | \- net.minidev:json-smart:jar:2.3:runtime
- [INFO] | | \- net.minidev:accessors-smart:jar:1.2:runtime
- [INFO] | | \- org.ow2.asm:asm:jar:5.0.4:runtime
+ [INFO] | +- com.jayway.jsonpath:json-path:jar:2.7.0:runtime
+ [INFO] | | \- net.minidev:json-smart:jar:2.4.7:runtime
+ [INFO] | | \- net.minidev:accessors-smart:jar:2.4.7:runtime
+ [INFO] | | \- org.ow2.asm:asm:jar:9.1:runtime
[INFO] | +- commons-codec:commons-codec:jar:1.15:runtime
- [INFO] | \- commons-io:commons-io:jar:2.11.0:compile
+ [INFO] | \- commons-io:commons-io:jar:2.11.0:runtime
Dependencies that are not needed for how we use Calcite right now.
@@ -107,11 +108,6 @@ under the License.
<groupId>net.hydromatic</groupId>
<artifactId>aggdesigner-algorithm</artifactId>
</exclusion>
- <!-- Excluding org.apiguardian:apiguardian-api v1.1.0 because org.junit.jupiter:junit-jupiter contains newer version v1.1.2 -->
- <exclusion>
- <groupId>org.apiguardian</groupId>
- <artifactId>apiguardian-api</artifactId>
- </exclusion>
<exclusion>
<groupId>com.google.errorprone</groupId>
<artifactId>error_prone_annotations</artifactId>
@@ -128,6 +124,26 @@ under the License.
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib-jdk8</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>org.codehaus.janino</groupId>
+ <artifactId>janino</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.codehaus.janino</groupId>
+ <artifactId>commons-compiler</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.httpcomponents.client5</groupId>
+ <artifactId>httpclient5</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.httpcomponents.core5</groupId>
+ <artifactId>httpcore5</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.checkerframework</groupId>
+ <artifactId>checker-qual</artifactId>
+ </exclusion>
</exclusions>
</dependency>
diff --git a/flink-table/flink-table-planner/pom.xml b/flink-table/flink-table-planner/pom.xml
index 70d8157a7e5..9dd0857a603 100644
--- a/flink-table/flink-table-planner/pom.xml
+++ b/flink-table/flink-table-planner/pom.xml
@@ -118,16 +118,10 @@ under the License.
</exclusion>
</exclusions>
</dependency>
- <!-- Use newer version v1.1.2 as in Calcite because org.junit.jupiter:junit-jupiter contains it -->
- <dependency>
- <artifactId>apiguardian-api</artifactId>
- <groupId>org.apiguardian</groupId>
- <version>1.1.2</version>
- </dependency>
<dependency>
<groupId>org.checkerframework</groupId>
<artifactId>checker-qual</artifactId>
- <version>3.10.0</version>
+ <version>3.12.0</version>
<optional>${flink.markBundledAsOptional}</optional>
</dependency>
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Filter.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Filter.java
deleted file mode 100644
index f1b9bf6b938..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Filter.java
+++ /dev/null
@@ -1,212 +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 org.apache.calcite.config.CalciteSystemProperty;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptCost;
-import org.apache.calcite.plan.RelOptPlanner;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelInput;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelWriter;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.calcite.rel.hint.Hintable;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rel.metadata.RelMdUtil;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.rex.RexChecker;
-import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.rex.RexOver;
-import org.apache.calcite.rex.RexProgram;
-import org.apache.calcite.rex.RexShuttle;
-import org.apache.calcite.rex.RexUtil;
-import org.apache.calcite.util.Litmus;
-import org.apiguardian.api.API;
-import org.checkerframework.checker.nullness.qual.EnsuresNonNullIf;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.List;
-import java.util.Objects;
-
-import static java.util.Objects.requireNonNull;
-
-/**
- * Relational expression that iterates over its input and returns elements for which <code>condition
- * </code> evaluates to <code>true</code>.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- *
- * <p>If the condition allows nulls, then a null value is treated the same as false.
- *
- * @see org.apache.calcite.rel.logical.LogicalFilter
- */
-public abstract class Filter extends SingleRel implements Hintable {
- // ~ Instance fields --------------------------------------------------------
-
- protected final RexNode condition;
-
- protected final ImmutableList<RelHint> hints;
-
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a filter.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param traits the traits of this rel
- * @param hints Hints for this node
- * @param child input relational expression
- * @param condition boolean expression which determines whether a row is allowed to pass
- */
- @SuppressWarnings("method.invocation.invalid")
- protected Filter(
- RelOptCluster cluster,
- RelTraitSet traits,
- List<RelHint> hints,
- RelNode child,
- RexNode condition) {
- super(cluster, traits, child);
- this.condition = requireNonNull(condition, "condition");
- assert RexUtil.isFlat(condition) : condition;
- // Too expensive for everyday use:
- assert !CalciteSystemProperty.DEBUG.value() || isValid(Litmus.THROW, null);
- this.hints = com.google.common.collect.ImmutableList.copyOf(hints);
- }
-
- /**
- * Creates a filter.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param traits the traits of this rel
- * @param child input relational expression
- * @param condition boolean expression which determines whether a row is allowed to pass
- */
- protected Filter(RelOptCluster cluster, RelTraitSet traits, RelNode child, RexNode condition) {
- this(cluster, traits, com.google.common.collect.ImmutableList.of(), child, condition);
- }
-
- /** Creates a Filter by parsing serialized output. */
- protected Filter(RelInput input) {
- this(
- input.getCluster(),
- input.getTraitSet(),
- input.getInput(),
- requireNonNull(input.getExpression("condition"), "condition"));
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public final RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return copy(traitSet, sole(inputs), getCondition());
- }
-
- public abstract Filter copy(RelTraitSet traitSet, RelNode input, RexNode condition);
-
- @Override
- public RelNode accept(RexShuttle shuttle) {
- RexNode condition = shuttle.apply(this.condition);
- if (this.condition == condition) {
- return this;
- }
- return copy(traitSet, getInput(), condition);
- }
-
- public RexNode getCondition() {
- return condition;
- }
-
- /** Returns whether this Filter contains any windowed-aggregate functions. */
- public final boolean containsOver() {
- return RexOver.containsOver(condition);
- }
-
- @Override
- public boolean isValid(Litmus litmus, Context context) {
- if (RexUtil.isNullabilityCast(getCluster().getTypeFactory(), condition)) {
- return litmus.fail("Cast for just nullability not allowed");
- }
- final RexChecker checker = new RexChecker(getInput().getRowType(), context, litmus);
- condition.accept(checker);
- if (checker.getFailureCount() > 0) {
- return litmus.fail(null);
- }
- return litmus.succeed();
- }
-
- @Override
- public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
- double dRows = mq.getRowCount(this);
- double dCpu = mq.getRowCount(getInput());
- double dIo = 0;
- return planner.getCostFactory().makeCost(dRows, dCpu, dIo);
- }
-
- @Override
- public double estimateRowCount(RelMetadataQuery mq) {
- return RelMdUtil.estimateFilteredRows(getInput(), condition, mq);
- }
-
- @Deprecated // to be removed before 2.0
- public static double estimateFilteredRows(RelNode child, RexProgram program) {
- final RelMetadataQuery mq = child.getCluster().getMetadataQuery();
- return RelMdUtil.estimateFilteredRows(child, program, mq);
- }
-
- @Deprecated // to be removed before 2.0
- public static double estimateFilteredRows(RelNode child, RexNode condition) {
- final RelMetadataQuery mq = child.getCluster().getMetadataQuery();
- return RelMdUtil.estimateFilteredRows(child, condition, mq);
- }
-
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- return super.explainTerms(pw).item("condition", condition);
- }
-
- @API(since = "1.24", status = API.Status.INTERNAL)
- @EnsuresNonNullIf(expression = "#1", result = true)
- protected boolean deepEquals0(@Nullable Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null || getClass() != obj.getClass()) {
- return false;
- }
- Filter o = (Filter) obj;
- return traitSet.equals(o.traitSet)
- && hints.equals(o.hints)
- && input.deepEquals(o.input)
- && condition.equals(o.condition)
- && getRowType().equalsSansFieldNames(o.getRowType());
- }
-
- @API(since = "1.24", status = API.Status.INTERNAL)
- protected int deepHashCode0() {
- return Objects.hash(traitSet, hints, input.deepHashCode(), condition);
- }
-
- @Override
- public ImmutableList<RelHint> getHints() {
- return hints;
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Intersect.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Intersect.java
deleted file mode 100644
index 9029903f50a..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Intersect.java
+++ /dev/null
@@ -1,77 +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 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.hint.RelHint;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.sql.SqlKind;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Relational expression that returns the intersection of the rows of its inputs.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- *
- * <p>If "all" is true, performs then multiset intersection; otherwise, performs set set
- * intersection (implying no duplicates in the results).
- */
-public abstract class Intersect extends SetOp {
- /** Creates an Intersect. */
- protected Intersect(
- RelOptCluster cluster,
- RelTraitSet traits,
- List<RelHint> hints,
- List<RelNode> inputs,
- boolean all) {
- super(cluster, traits, hints, inputs, SqlKind.INTERSECT, all);
- }
-
- /** Creates an Intersect. */
- public Intersect(RelOptCluster cluster, RelTraitSet traits, List<RelNode> inputs, boolean all) {
- this(cluster, traits, Collections.emptyList(), inputs, all);
- }
-
- /** Creates an Intersect by parsing serialized output. */
- protected Intersect(RelInput input) {
- super(input);
- }
-
- @Override
- public double estimateRowCount(RelMetadataQuery mq) {
- // REVIEW jvs 30-May-2005: I just pulled this out of a hat.
- double dRows = Double.MAX_VALUE;
- for (RelNode input : inputs) {
- Double rowCount = mq.getRowCount(input);
- if (rowCount == null) {
- // Assume this input does not reduce row count
- continue;
- }
- dRows = Math.min(dRows, rowCount);
- }
- dRows *= 0.25;
- return dRows;
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Minus.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Minus.java
deleted file mode 100644
index 15196253055..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Minus.java
+++ /dev/null
@@ -1,69 +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 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.hint.RelHint;
-import org.apache.calcite.rel.metadata.RelMdUtil;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.sql.SqlKind;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Relational expression that returns the rows of its first input minus any matching rows from its
- * other inputs.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- *
- * <p>Corresponds to the SQL {@code EXCEPT} operator.
- *
- * <p>If "all" is true, then multiset subtraction is performed; otherwise, set subtraction is
- * performed (implying no duplicates in the results).
- */
-public abstract class Minus extends SetOp {
-
- protected Minus(
- RelOptCluster cluster,
- RelTraitSet traits,
- List<RelHint> hints,
- List<RelNode> inputs,
- boolean all) {
- super(cluster, traits, hints, inputs, SqlKind.EXCEPT, all);
- }
-
- public Minus(RelOptCluster cluster, RelTraitSet traits, List<RelNode> inputs, boolean all) {
- this(cluster, traits, Collections.emptyList(), inputs, all);
- }
-
- /** Creates a Minus by parsing serialized output. */
- protected Minus(RelInput input) {
- super(input);
- }
-
- @Override
- public double estimateRowCount(RelMetadataQuery mq) {
- return RelMdUtil.getMinusRowCount(mq, this);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/SetOp.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/SetOp.java
deleted file mode 100644
index 93fe1131168..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/SetOp.java
+++ /dev/null
@@ -1,159 +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 org.apache.calcite.linq4j.Ord;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptUtil;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.AbstractRelNode;
-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.type.RelDataType;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.util.Util;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-/**
- * <code>SetOp</code> is an abstract base for relational set operators such as UNION, MINUS (aka
- * EXCEPT), and INTERSECT.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public abstract class SetOp extends AbstractRelNode implements Hintable {
- // ~ Instance fields --------------------------------------------------------
-
- protected ImmutableList<RelNode> inputs;
- public final SqlKind kind;
- public final boolean all;
- protected final ImmutableList<RelHint> hints;
-
- // ~ Constructors -----------------------------------------------------------
-
- /** Creates a SetOp. */
- protected SetOp(
- RelOptCluster cluster,
- RelTraitSet traits,
- List<RelHint> hints,
- List<RelNode> inputs,
- SqlKind kind,
- boolean all) {
- super(cluster, traits);
- com.google.common.base.Preconditions.checkArgument(
- kind == SqlKind.UNION || kind == SqlKind.INTERSECT || kind == SqlKind.EXCEPT);
- this.kind = kind;
- this.inputs = ImmutableList.copyOf(inputs);
- this.all = all;
- this.hints = ImmutableList.copyOf(hints);
- }
-
- /** Creates a SetOp. */
- protected SetOp(
- RelOptCluster cluster,
- RelTraitSet traits,
- List<RelNode> inputs,
- SqlKind kind,
- boolean all) {
- this(cluster, traits, Collections.emptyList(), inputs, kind, all);
- }
-
- /** Creates a SetOp by parsing serialized output. */
- protected SetOp(RelInput input) {
- this(
- input.getCluster(),
- input.getTraitSet(),
- Collections.emptyList(),
- input.getInputs(),
- SqlKind.UNION,
- input.getBoolean("all", false));
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- public abstract SetOp copy(RelTraitSet traitSet, List<RelNode> inputs, boolean all);
-
- @Override
- public SetOp copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return copy(traitSet, inputs, all);
- }
-
- @Override
- public void replaceInput(int ordinalInParent, RelNode p) {
- final List<RelNode> newInputs = new ArrayList<>(inputs);
- newInputs.set(ordinalInParent, p);
- inputs = ImmutableList.copyOf(newInputs);
- recomputeDigest();
- }
-
- @Override
- public List<RelNode> getInputs() {
- return inputs;
- }
-
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- super.explainTerms(pw);
- for (Ord<RelNode> ord : Ord.zip(inputs)) {
- pw.input("input#" + ord.i, ord.e);
- }
- return pw.item("all", all);
- }
-
- @Override
- protected RelDataType deriveRowType() {
- final List<RelDataType> inputRowTypes = Util.transform(inputs, RelNode::getRowType);
- final RelDataType rowType = getCluster().getTypeFactory().leastRestrictive(inputRowTypes);
- if (rowType == null) {
- throw new IllegalArgumentException(
- "Cannot compute compatible row type "
- + "for arguments to set op: "
- + Util.sepList(inputRowTypes, ", "));
- }
- return rowType;
- }
-
- /**
- * Returns whether all the inputs of this set operator have the same row type as its output row.
- *
- * @param compareNames Whether column names are important in the homogeneity comparison
- * @return Whether all the inputs of this set operator have the same row type as its output row
- */
- public boolean isHomogeneous(boolean compareNames) {
- RelDataType unionType = getRowType();
- for (RelNode input : getInputs()) {
- if (!RelOptUtil.areRowTypesEqual(input.getRowType(), unionType, compareNames)) {
- return false;
- }
- }
- return true;
- }
-
- @Override
- public com.google.common.collect.ImmutableList<RelHint> getHints() {
- return hints;
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Snapshot.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Snapshot.java
index 531f69345dc..e08b354a8ec 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Snapshot.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Snapshot.java
@@ -98,6 +98,7 @@ public abstract class Snapshot extends SingleRel implements Hintable {
public abstract Snapshot copy(RelTraitSet traitSet, RelNode input, RexNode period);
+ @Override
public RelNode accept(RexShuttle shuttle) {
RexNode condition = shuttle.apply(this.period);
if (this.period == condition) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Sort.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Sort.java
deleted file mode 100644
index f8408b91f94..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Sort.java
+++ /dev/null
@@ -1,291 +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 org.apache.calcite.linq4j.Ord;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptCost;
-import org.apache.calcite.plan.RelOptPlanner;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollation;
-import org.apache.calcite.rel.RelCollationTraitDef;
-import org.apache.calcite.rel.RelFieldCollation;
-import org.apache.calcite.rel.RelInput;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelWriter;
-import org.apache.calcite.rel.SingleRel;
-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.rex.RexLiteral;
-import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.rex.RexShuttle;
-import org.apache.calcite.util.Util;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-
-/**
- * Relational expression that imposes a particular sort order on its input without otherwise
- * changing its content.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public abstract class Sort extends SingleRel implements Hintable {
- // ~ Instance fields --------------------------------------------------------
-
- public final RelCollation collation;
- public final @Nullable RexNode offset;
- public final @Nullable RexNode fetch;
- protected final ImmutableList<RelHint> hints;
-
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a Sort.
- *
- * @param cluster Cluster this relational expression belongs to
- * @param traits Traits
- * @param child input relational expression
- * @param collation array of sort specifications
- */
- protected Sort(
- RelOptCluster cluster, RelTraitSet traits, RelNode child, RelCollation collation) {
- this(cluster, traits, Collections.emptyList(), child, collation, null, null);
- }
-
- /**
- * Creates a Sort.
- *
- * @param cluster Cluster this relational expression belongs to
- * @param traits Traits
- * @param child input relational expression
- * @param collation array of sort specifications
- * @param offset Expression for number of rows to discard before returning first row
- * @param fetch Expression for number of rows to fetch
- */
- public Sort(
- RelOptCluster cluster,
- RelTraitSet traits,
- RelNode child,
- RelCollation collation,
- RexNode offset,
- RexNode fetch) {
- this(cluster, traits, Collections.emptyList(), child, collation, offset, fetch);
- }
-
- /**
- * Creates a Sort.
- *
- * @param cluster Cluster this relational expression belongs to
- * @param traits Traits
- * @param hints Hints for this node
- * @param child input relational expression
- * @param collation array of sort specifications
- * @param offset Expression for number of rows to discard before returning first row
- * @param fetch Expression for number of rows to fetch
- */
- protected Sort(
- RelOptCluster cluster,
- RelTraitSet traits,
- List<RelHint> hints,
- RelNode child,
- RelCollation collation,
- @Nullable RexNode offset,
- @Nullable RexNode fetch) {
- super(cluster, traits, child);
- this.collation = collation;
- this.offset = offset;
- this.fetch = fetch;
- this.hints = ImmutableList.copyOf(hints);
-
- assert traits.containsIfApplicable(collation)
- : "traits=" + traits + ", collation=" + collation;
- assert !(fetch == null && offset == null && collation.getFieldCollations().isEmpty())
- : "trivial sort";
- }
-
- /** Creates a Sort by parsing serialized output. */
- protected Sort(RelInput input) {
- this(
- input.getCluster(),
- input.getTraitSet().plus(input.getCollation()),
- input.getInput(),
- RelCollationTraitDef.INSTANCE.canonize(input.getCollation()),
- input.getExpression("offset"),
- input.getExpression("fetch"));
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public final Sort copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return copy(traitSet, sole(inputs), collation, offset, fetch);
- }
-
- public final Sort copy(RelTraitSet traitSet, RelNode newInput, RelCollation newCollation) {
- return copy(traitSet, newInput, newCollation, offset, fetch);
- }
-
- public abstract Sort copy(
- RelTraitSet traitSet,
- RelNode newInput,
- RelCollation newCollation,
- @Nullable RexNode offset,
- @Nullable RexNode fetch);
-
- /**
- * {@inheritDoc}
- *
- * <p>The CPU cost of a Sort has three main cases:
- *
- * <ul>
- * <li>If {@code fetch} is zero, CPU cost is zero; otherwise,
- * <li>if the sort keys are empty, we don't need to sort, only step over the rows, and
- * therefore the CPU cost is {@code min(fetch + offset, inputRowCount) * bytesPerRow};
- * otherwise
- * <li>we need to read and sort {@code inputRowCount} rows, with at most {@code min(fetch +
- * offset, inputRowCount)} of them in the sort data structure at a time, giving a CPU cost
- * of {@code inputRowCount * log(min(fetch + offset, inputRowCount)) * bytesPerRow}.
- * </ul>
- *
- * <p>The cost model factors in row width via {@code bytesPerRow}, because sorts need to move
- * rows around, not just compare them; by making the cost higher if rows are wider, we
- * discourage pushing a Project through a Sort. We assume that each field is 4 bytes, and we add
- * 3 'virtual fields' to represent the per-row overhead. Thus a 1-field row is (3 + 1) * 4 = 16
- * bytes; a 5-field row is (3 + 5) * 4 = 32 bytes.
- *
- * <p>The cost model does not consider a 5-field sort to be more expensive than, say, a 2-field
- * sort, because both sorts will compare just one field most of the time.
- */
- @Override
- public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
- final double offsetValue = Util.first(doubleValue(offset), 0d);
- assert offsetValue >= 0 : "offset should not be negative:" + offsetValue;
-
- final double inCount = mq.getRowCount(input);
- @Nullable Double fetchValue = doubleValue(fetch);
- final double readCount;
- if (fetchValue == null) {
- readCount = inCount;
- } else if (fetchValue <= 0) {
- // Case 1. Read zero rows from input, therefore CPU cost is zero.
- return planner.getCostFactory().makeCost(inCount, 0, 0);
- } else {
- readCount = Math.min(inCount, offsetValue + fetchValue);
- }
-
- final double bytesPerRow = (3 + getRowType().getFieldCount()) * 4;
-
- final double cpu;
- if (collation.getFieldCollations().isEmpty()) {
- // Case 2. If sort keys are empty, CPU cost is cheaper because we are just
- // stepping over the first "readCount" rows, rather than sorting all
- // "inCount" them. (Presumably we are applying FETCH and/or OFFSET,
- // otherwise this Sort is a no-op.)
- cpu = readCount * bytesPerRow;
- } else {
- // Case 3. Read and sort all "inCount" rows, keeping "readCount" in the
- // sort data structure at a time.
- cpu = Util.nLogM(inCount, readCount) * bytesPerRow;
- }
- return planner.getCostFactory().makeCost(readCount, cpu, 0);
- }
-
- @Override
- public RelNode accept(RexShuttle shuttle) {
- RexNode offset = shuttle.apply(this.offset);
- RexNode fetch = shuttle.apply(this.fetch);
- List<RexNode> originalSortExps = getSortExps();
- List<RexNode> sortExps = shuttle.apply(originalSortExps);
- assert sortExps == originalSortExps
- : "Sort node does not support modification of input field expressions."
- + " Old expressions: "
- + originalSortExps
- + ", new ones: "
- + sortExps;
- if (offset == this.offset && fetch == this.fetch) {
- return this;
- }
- return copy(traitSet, getInput(), collation, offset, fetch);
- }
-
- @Override
- public boolean isEnforcer() {
- return offset == null && fetch == null && collation.getFieldCollations().size() > 0;
- }
-
- /**
- * Returns the array of {@link RelFieldCollation}s asked for by the sort specification, from
- * most significant to least significant.
- *
- * <p>See also {@link RelMetadataQuery#collations(RelNode)}, which lists all known collations.
- * For example, <code>ORDER BY time_id</code> might also be sorted by <code>the_year, the_month
- * </code> because of a known monotonicity constraint among the columns. {@code getCollation}
- * would return <code>[time_id]</code> and {@code collations} would return <code>
- * [ [time_id], [the_year, the_month] ]</code>.
- */
- public RelCollation getCollation() {
- return collation;
- }
-
- /** Returns the sort expressions. */
- public List<RexNode> getSortExps() {
- //noinspection StaticPseudoFunctionalStyleMethod
- return Util.transform(
- collation.getFieldCollations(),
- field ->
- getCluster()
- .getRexBuilder()
- .makeInputRef(
- input, Objects.requireNonNull(field).getFieldIndex()));
- }
-
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- super.explainTerms(pw);
- if (pw.nest()) {
- pw.item("collation", collation);
- } else {
- for (Ord<RexNode> ord : Ord.zip(getSortExps())) {
- pw.item("sort" + ord.i, ord.e);
- }
- for (Ord<RelFieldCollation> ord : Ord.zip(collation.getFieldCollations())) {
- pw.item("dir" + ord.i, ord.e.shortString());
- }
- }
- pw.itemIf("offset", offset, offset != null);
- pw.itemIf("fetch", fetch, fetch != null);
- return pw;
- }
-
- @Override
- public ImmutableList<RelHint> getHints() {
- return hints;
- }
-
- /** Returns the double value of a node if it is a literal, otherwise null. */
- private static @Nullable Double doubleValue(@Nullable RexNode r) {
- return r instanceof RexLiteral ? ((RexLiteral) r).getValueAs(Double.class) : null;
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Union.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Union.java
deleted file mode 100644
index f4099ae1f9a..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Union.java
+++ /dev/null
@@ -1,79 +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 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.hint.RelHint;
-import org.apache.calcite.rel.metadata.RelMdUtil;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.sql.SqlKind;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Relational expression that returns the union of the rows of its inputs, optionally eliminating
- * duplicates.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- *
- * <p>Corresponds to SQL {@code UNION} and {@code UNION ALL}.
- */
-public abstract class Union extends SetOp {
- // ~ Constructors -----------------------------------------------------------
-
- protected Union(
- RelOptCluster cluster,
- RelTraitSet traits,
- List<RelHint> hints,
- List<RelNode> inputs,
- boolean all) {
- super(cluster, traits, hints, inputs, SqlKind.UNION, all);
- }
-
- protected Union(RelOptCluster cluster, RelTraitSet traits, List<RelNode> inputs, boolean all) {
- super(cluster, traits, Collections.emptyList(), inputs, SqlKind.UNION, all);
- }
-
- /** Creates a Union by parsing serialized output. */
- protected Union(RelInput input) {
- super(input);
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public double estimateRowCount(RelMetadataQuery mq) {
- double dRows = RelMdUtil.getUnionAllRowCount(mq, this);
- if (!all) {
- dRows *= 0.5;
- }
- return dRows;
- }
-
- @Deprecated // to be removed before 2.0
- public static double estimateRowCount(RelNode rel) {
- final RelMetadataQuery mq = rel.getCluster().getMetadataQuery();
- return RelMdUtil.getUnionAllRowCount(mq, (Union) rel);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Values.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Values.java
deleted file mode 100644
index 8032241426d..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Values.java
+++ /dev/null
@@ -1,248 +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 org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptCost;
-import org.apache.calcite.plan.RelOptPlanner;
-import org.apache.calcite.plan.RelOptRule;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.AbstractRelNode;
-import org.apache.calcite.rel.RelInput;
-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.rel.type.RelDataTypeField;
-import org.apache.calcite.rex.RexDigestIncludeType;
-import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.sql.SqlExplainLevel;
-import org.apache.calcite.sql.type.SqlTypeUtil;
-import org.apache.calcite.util.Pair;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.function.Predicate;
-import java.util.stream.Collectors;
-
-/**
- * Relational expression whose value is a sequence of zero or more literal row values.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public abstract class Values extends AbstractRelNode implements Hintable {
-
- public static final Predicate<? super Values> IS_EMPTY_J = Values::isEmpty;
-
- protected final ImmutableList<RelHint> hints;
-
- @SuppressWarnings("Guava")
- @Deprecated // to be removed before 2.0
- public static final com.google.common.base.Predicate<? super Values> IS_EMPTY = Values::isEmpty;
-
- @SuppressWarnings("Guava")
- @Deprecated // to be removed before 2.0
- public static final com.google.common.base.Predicate<? super Values> IS_NOT_EMPTY =
- Values::isNotEmpty;
-
- // ~ Instance fields --------------------------------------------------------
-
- public final ImmutableList<ImmutableList<RexLiteral>> tuples;
-
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a new Values.
- *
- * <p>Note that tuples passed in become owned by this rel (without a deep copy), so caller must
- * not modify them after this call, otherwise bad things will happen.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param hints Hints for this node
- * @param rowType Row type for tuples produced by this rel
- * @param tuples 2-dimensional array of tuple values to be produced; outer list contains tuples;
- * each inner list is one tuple; all tuples must be of same length, conforming to rowType
- */
- protected Values(
- RelOptCluster cluster,
- List<RelHint> hints,
- RelDataType rowType,
- ImmutableList<ImmutableList<RexLiteral>> tuples,
- RelTraitSet traits) {
- super(cluster, traits);
- this.rowType = rowType;
- this.tuples = tuples;
- this.hints = ImmutableList.copyOf(hints);
- assert assertRowType();
- }
-
- /**
- * Creates a new Values.
- *
- * <p>Note that tuples passed in become owned by this rel (without a deep copy), so caller must
- * not modify them after this call, otherwise bad things will happen.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param rowType Row type for tuples produced by this rel
- * @param tuples 2-dimensional array of tuple values to be produced; outer list contains tuples;
- * each inner list is one tuple; all tuples must be of same length, conforming to rowType
- */
- protected Values(
- RelOptCluster cluster,
- RelDataType rowType,
- com.google.common.collect.ImmutableList<
- com.google.common.collect.ImmutableList<RexLiteral>>
- tuples,
- RelTraitSet traits) {
- this(cluster, Collections.emptyList(), rowType, tuples, traits);
- }
-
- /** Creates a Values by parsing serialized output. */
- protected Values(RelInput input) {
- this(
- input.getCluster(),
- input.getRowType("type"),
- input.getTuples("tuples"),
- input.getTraitSet());
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- /**
- * Predicate, to be used when defining an operand of a {@link RelOptRule}, that returns true if
- * a Values contains zero tuples.
- *
- * <p>This is the conventional way to represent an empty relational expression. There are
- * several rules that recognize empty relational expressions and prune away that section of the
- * tree.
- */
- public static boolean isEmpty(Values values) {
- return values.getTuples().isEmpty();
- }
-
- /**
- * Predicate, to be used when defining an operand of a {@link RelOptRule}, that returns true if
- * a Values contains one or more tuples.
- *
- * <p>This is the conventional way to represent an empty relational expression. There are
- * several rules that recognize empty relational expressions and prune away that section of the
- * tree.
- */
- public static boolean isNotEmpty(Values values) {
- return !isEmpty(values);
- }
-
- public ImmutableList<ImmutableList<RexLiteral>> getTuples(RelInput input) {
- return input.getTuples("tuples");
- }
-
- /** Returns the rows of literals represented by this Values relational expression. */
- public ImmutableList<ImmutableList<RexLiteral>> getTuples() {
- return tuples;
- }
-
- /** Returns true if all tuples match rowType; otherwise, assert on mismatch. */
- private boolean assertRowType() {
- RelDataType rowType = getRowType();
- for (List<RexLiteral> tuple : tuples) {
- assert tuple.size() == rowType.getFieldCount();
- for (Pair<RexLiteral, RelDataTypeField> pair :
- Pair.zip(tuple, rowType.getFieldList())) {
- RexLiteral literal = pair.left;
- RelDataType fieldType = pair.right.getType();
-
- // TODO jvs 19-Feb-2006: strengthen this a bit. For example,
- // overflow, rounding, and padding/truncation must already have
- // been dealt with.
- if (!RexLiteral.isNullLiteral(literal)) {
- assert SqlTypeUtil.canAssignFrom(fieldType, literal.getType())
- : "to " + fieldType + " from " + literal;
- }
- }
- }
- return true;
- }
-
- @Override
- protected RelDataType deriveRowType() {
- assert rowType != null : "rowType must not be null for " + this;
- return rowType;
- }
-
- @Override
- public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
- double dRows = mq.getRowCount(this);
-
- // Assume CPU is negligible since values are precomputed.
- double dCpu = 1;
- double dIo = 0;
- return planner.getCostFactory().makeCost(dRows, dCpu, dIo);
- }
-
- // implement RelNode
- @Override
- public double estimateRowCount(RelMetadataQuery mq) {
- return tuples.size();
- }
-
- // implement RelNode
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- // A little adapter just to get the tuples to come out
- // with curly brackets instead of square brackets. Plus
- // more whitespace for readability.
- RelDataType rowType = getRowType();
- RelWriter relWriter =
- super.explainTerms(pw)
- // For rel digest, include the row type since a rendered
- // literal may leave the type ambiguous (e.g. "null").
- .itemIf(
- "type",
- rowType,
- pw.getDetailLevel() == SqlExplainLevel.DIGEST_ATTRIBUTES)
- .itemIf("type", rowType.getFieldList(), pw.nest());
- if (pw.nest()) {
- pw.item("tuples", tuples);
- } else {
- pw.item(
- "tuples",
- tuples.stream()
- .map(
- row ->
- row.stream()
- .map(
- lit ->
- lit.computeDigest(
- RexDigestIncludeType
- .NO_TYPE))
- .collect(Collectors.joining(", ", "{ ", " }")))
- .collect(Collectors.joining(", ", "[", "]")));
- }
- return relWriter;
- }
-
- @Override
- public ImmutableList<RelHint> getHints() {
- return hints;
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Window.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Window.java
deleted file mode 100644
index 5b797d42186..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/core/Window.java
+++ /dev/null
@@ -1,484 +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 org.apache.calcite.linq4j.Ord;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptCost;
-import org.apache.calcite.plan.RelOptPlanner;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollation;
-import org.apache.calcite.rel.RelCollations;
-import org.apache.calcite.rel.RelFieldCollation;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelWriter;
-import org.apache.calcite.rel.SingleRel;
-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.rex.RexCall;
-import org.apache.calcite.rex.RexChecker;
-import org.apache.calcite.rex.RexFieldCollation;
-import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.rex.RexLocalRef;
-import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.rex.RexSlot;
-import org.apache.calcite.rex.RexWindowBound;
-import org.apache.calcite.sql.SqlAggFunction;
-import org.apache.calcite.util.ImmutableBitSet;
-import org.apache.calcite.util.ImmutableIntList;
-import org.apache.calcite.util.Litmus;
-import org.apache.calcite.util.Util;
-import org.checkerframework.checker.nullness.qual.Nullable;
-import org.checkerframework.checker.nullness.qual.RequiresNonNull;
-
-import java.util.AbstractList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-
-/**
- * A relational expression representing a set of window aggregates.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- *
- * <p>A Window can handle several window aggregate functions, over several partitions, with pre- and
- * post-expressions, and an optional post-filter. Each of the partitions is defined by a partition
- * key (zero or more columns) and a range (logical or physical). The partitions expect the data to
- * be sorted correctly on input to the relational expression.
- *
- * <p>Each {@link Window.Group} has a set of {@link org.apache.calcite.rex.RexOver} objects.
- *
- * <p>Created by {@link org.apache.calcite.rel.rules.ProjectToWindowRule}.
- */
-public abstract class Window extends SingleRel implements Hintable {
- public final ImmutableList<Group> groups;
- public final ImmutableList<RexLiteral> constants;
- protected final ImmutableList<RelHint> hints;
-
- /**
- * Creates a window relational expression.
- *
- * @param cluster Cluster
- * @param traitSet Trait set
- * @param hints Hints for this node
- * @param input Input relational expression
- * @param constants List of constants that are additional inputs
- * @param rowType Output row type
- * @param groups Windows
- */
- protected Window(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- RelNode input,
- List<RexLiteral> constants,
- RelDataType rowType,
- List<Group> groups) {
- super(cluster, traitSet, input);
- this.constants = ImmutableList.copyOf(constants);
- assert rowType != null;
- this.rowType = rowType;
- this.groups = ImmutableList.copyOf(groups);
- this.hints = ImmutableList.copyOf(hints);
- }
-
- /**
- * Creates a window relational expression.
- *
- * @param cluster Cluster
- * @param traitSet Trait set
- * @param input Input relational expression
- * @param constants List of constants that are additional inputs
- * @param rowType Output row type
- * @param groups Windows
- */
- public Window(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelNode input,
- List<RexLiteral> constants,
- RelDataType rowType,
- List<Group> groups) {
- this(cluster, traitSet, Collections.emptyList(), input, constants, rowType, groups);
- }
-
- @Override
- public boolean isValid(Litmus litmus, RelNode.Context context) {
- // In the window specifications, an aggregate call such as
- // 'SUM(RexInputRef #10)' refers to expression #10 of inputProgram.
- // (Not its projections.)
- final RelDataType childRowType = getInput().getRowType();
-
- final int childFieldCount = childRowType.getFieldCount();
- final int inputSize = childFieldCount + constants.size();
- final List<RelDataType> inputTypes =
- new AbstractList<RelDataType>() {
- @Override
- public RelDataType get(int index) {
- return index < childFieldCount
- ? childRowType.getFieldList().get(index).getType()
- : constants.get(index - childFieldCount).getType();
- }
-
- @Override
- public int size() {
- return inputSize;
- }
- };
-
- final RexChecker checker = new RexChecker(inputTypes, context, litmus);
- int count = 0;
- for (Group group : groups) {
- for (RexWinAggCall over : group.aggCalls) {
- ++count;
- if (!checker.isValid(over)) {
- return litmus.fail(null);
- }
- }
- }
- if (count == 0) {
- return litmus.fail("empty");
- }
- return litmus.succeed();
- }
-
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- super.explainTerms(pw);
- for (Ord<Group> window : Ord.zip(groups)) {
- pw.item("window#" + window.i, window.e.toString());
- }
- return pw;
- }
-
- public static ImmutableIntList getProjectOrdinals(final List<RexNode> exprs) {
- return ImmutableIntList.copyOf(
- new AbstractList<Integer>() {
- @Override
- public Integer get(int index) {
- return ((RexSlot) exprs.get(index)).getIndex();
- }
-
- @Override
- public int size() {
- return exprs.size();
- }
- });
- }
-
- public static RelCollation getCollation(final List<RexFieldCollation> collations) {
- return RelCollations.of(
- new AbstractList<RelFieldCollation>() {
- @Override
- public RelFieldCollation get(int index) {
- final RexFieldCollation collation = collations.get(index);
- return new RelFieldCollation(
- ((RexLocalRef) collation.left).getIndex(),
- collation.getDirection(),
- collation.getNullDirection());
- }
-
- @Override
- public int size() {
- return collations.size();
- }
- });
- }
-
- /**
- * Returns constants that are additional inputs of current relation.
- *
- * @return constants that are additional inputs of current relation
- */
- public List<RexLiteral> getConstants() {
- return constants;
- }
-
- @Override
- public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
- // Cost is proportional to the number of rows and the number of
- // components (groups and aggregate functions). There is
- // no I/O cost.
- //
- // TODO #1. Add memory cost.
- // TODO #2. MIN and MAX have higher CPU cost than SUM and COUNT.
- final double rowsIn = mq.getRowCount(getInput());
- int count = groups.size();
- for (Group group : groups) {
- count += group.aggCalls.size();
- }
- return planner.getCostFactory().makeCost(rowsIn, rowsIn * count, 0);
- }
-
- /**
- * Group of windowed aggregate calls that have the same window specification.
- *
- * <p>The specification is defined by an upper and lower bound, and also has zero or more
- * partitioning columns.
- *
- * <p>A window is either logical or physical. A physical window is measured in terms of row
- * count. A logical window is measured in terms of rows within a certain distance from the
- * current sort key.
- *
- * <p>For example:
- *
- * <ul>
- * <li><code>ROWS BETWEEN 10 PRECEDING and 5 FOLLOWING</code> is a physical window with an
- * upper and lower bound;
- * <li><code>RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND UNBOUNDED
- * FOLLOWING</code> is a logical window with only a lower bound;
- * <li><code>RANGE INTERVAL '10' MINUTES PRECEDING</code> (which is equivalent to <code>
- * RANGE BETWEEN INTERVAL '10' MINUTES PRECEDING AND
- * CURRENT ROW</code>) is a logical window with an upper and lower bound.
- * </ul>
- */
- public static class Group {
- public final ImmutableBitSet keys;
- public final boolean isRows;
- public final RexWindowBound lowerBound;
- public final RexWindowBound upperBound;
- public final RelCollation orderKeys;
- private final String digest;
-
- /**
- * List of {@link Window.RexWinAggCall} objects, each of which is a call to a {@link
- * org.apache.calcite.sql.SqlAggFunction}.
- */
- public final ImmutableList<RexWinAggCall> aggCalls;
-
- public Group(
- ImmutableBitSet keys,
- boolean isRows,
- RexWindowBound lowerBound,
- RexWindowBound upperBound,
- RelCollation orderKeys,
- List<RexWinAggCall> aggCalls) {
- this.keys = Objects.requireNonNull(keys, "keys");
- this.isRows = isRows;
- this.lowerBound = Objects.requireNonNull(lowerBound, "lowerBound");
- this.upperBound = Objects.requireNonNull(upperBound, "upperBound");
- this.orderKeys = Objects.requireNonNull(orderKeys, "orderKeys");
- this.aggCalls = ImmutableList.copyOf(aggCalls);
- this.digest = computeString();
- }
-
- @Override
- public String toString() {
- return digest;
- }
-
- @RequiresNonNull({"keys", "orderKeys", "lowerBound", "upperBound", "aggCalls"})
- private String computeString() {
- final StringBuilder buf = new StringBuilder("window(");
- final int i = buf.length();
- if (!keys.isEmpty()) {
- buf.append("partition ");
- buf.append(keys);
- }
- if (!orderKeys.getFieldCollations().isEmpty()) {
- buf.append(buf.length() == i ? "order by " : " order by ");
- buf.append(orderKeys);
- }
- if (orderKeys.getFieldCollations().isEmpty()
- && lowerBound.isUnbounded()
- && lowerBound.isPreceding()
- && upperBound.isUnbounded()
- && upperBound.isFollowing()) {
- // skip bracket if no ORDER BY, and if bracket is the default,
- // "RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING",
- // which is equivalent to
- // "ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING"
- } else if (!orderKeys.getFieldCollations().isEmpty()
- && lowerBound.isUnbounded()
- && lowerBound.isPreceding()
- && upperBound.isCurrentRow()
- && !isRows) {
- // skip bracket if there is ORDER BY, and if bracket is the default,
- // "RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW",
- // which is NOT equivalent to
- // "ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"
- } else {
- buf.append(isRows ? " rows " : " range ");
- buf.append("between ");
- buf.append(lowerBound);
- buf.append(" and ");
- buf.append(upperBound);
- }
- if (!aggCalls.isEmpty()) {
- buf.append(buf.length() == i ? "aggs " : " aggs ");
- buf.append(aggCalls);
- }
- buf.append(")");
- return buf.toString();
- }
-
- @Override
- public boolean equals(@Nullable Object obj) {
- return this == obj || obj instanceof Group && this.digest.equals(((Group) obj).digest);
- }
-
- @Override
- public int hashCode() {
- return digest.hashCode();
- }
-
- public RelCollation collation() {
- return orderKeys;
- }
-
- /**
- * Returns if the window is guaranteed to have rows. This is useful to refine data type of
- * window aggregates. For instance sum(non-nullable) over (empty window) is NULL.
- *
- * @return true when the window is non-empty
- * @see org.apache.calcite.sql.SqlWindow#isAlwaysNonEmpty()
- * @see org.apache.calcite.sql.SqlOperatorBinding#getGroupCount()
- * @see
- * org.apache.calcite.sql.validate.SqlValidatorImpl#resolveWindow(org.apache.calcite.sql.SqlNode,
- * org.apache.calcite.sql.validate.SqlValidatorScope)
- */
- public boolean isAlwaysNonEmpty() {
- int lowerKey = lowerBound.getOrderKey();
- int upperKey = upperBound.getOrderKey();
- return lowerKey > -1 && lowerKey <= upperKey;
- }
-
- /** Presents a view of the {@link RexWinAggCall} list as a list of {@link AggregateCall}. */
- public List<AggregateCall> getAggregateCalls(Window windowRel) {
- final List<String> fieldNames =
- Util.skip(
- windowRel.getRowType().getFieldNames(),
- windowRel.getInput().getRowType().getFieldCount());
- return new AbstractList<AggregateCall>() {
- @Override
- public int size() {
- return aggCalls.size();
- }
-
- @Override
- public AggregateCall get(int index) {
- final RexWinAggCall aggCall = aggCalls.get(index);
- final SqlAggFunction op = (SqlAggFunction) aggCall.getOperator();
- return AggregateCall.create(
- op,
- aggCall.distinct,
- false,
- aggCall.ignoreNulls,
- getProjectOrdinals(aggCall.getOperands()),
- -1,
- null,
- RelCollations.EMPTY,
- aggCall.getType(),
- fieldNames.get(aggCall.ordinal));
- }
- };
- }
- }
-
- /**
- * A call to a windowed aggregate function.
- *
- * <p>Belongs to a {@link Window.Group}.
- *
- * <p>It's a bastard son of a {@link org.apache.calcite.rex.RexCall}; similar enough that it
- * gets visited by a {@link org.apache.calcite.rex.RexVisitor}, but it also has some extra data
- * members.
- */
- public static class RexWinAggCall extends RexCall {
- /** Ordinal of this aggregate within its partition. */
- public final int ordinal;
-
- /** Whether to eliminate duplicates before applying aggregate function. */
- public final boolean distinct;
-
- /** Whether to ignore nulls. */
- public final boolean ignoreNulls;
-
- @Deprecated // to be removed before 2.0
- public RexWinAggCall(
- SqlAggFunction aggFun,
- RelDataType type,
- List<RexNode> operands,
- int ordinal,
- boolean distinct) {
- this(aggFun, type, operands, ordinal, distinct, false);
- }
-
- /**
- * Creates a RexWinAggCall.
- *
- * @param aggFun Aggregate function
- * @param type Result type
- * @param operands Operands to call
- * @param ordinal Ordinal within its partition
- * @param distinct Eliminate duplicates before applying aggregate function
- */
- public RexWinAggCall(
- SqlAggFunction aggFun,
- RelDataType type,
- List<RexNode> operands,
- int ordinal,
- boolean distinct,
- boolean ignoreNulls) {
- super(type, aggFun, operands);
- this.ordinal = ordinal;
- this.distinct = distinct;
- this.ignoreNulls = ignoreNulls;
- }
-
- @Override
- public boolean equals(@Nullable Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- RexWinAggCall that = (RexWinAggCall) o;
- return ordinal == that.ordinal
- && distinct == that.distinct
- && ignoreNulls == that.ignoreNulls;
- }
-
- @Override
- public int hashCode() {
- if (hash == 0) {
- hash = Objects.hash(super.hashCode(), ordinal, distinct, ignoreNulls);
- }
- return hash;
- }
-
- @Override
- public RexCall clone(RelDataType type, List<RexNode> operands) {
- return super.clone(type, operands);
- }
- }
-
- @Override
- public com.google.common.collect.ImmutableList<RelHint> getHints() {
- return hints;
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/hint/HintPredicates.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/hint/HintPredicates.java
deleted file mode 100644
index 7081c68041a..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/hint/HintPredicates.java
+++ /dev/null
@@ -1,140 +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.hint;
-
-/**
- * A collection of hint predicates.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade *
- * the latest calcite.
- */
-public abstract class HintPredicates {
- /**
- * A hint predicate that indicates a hint can only be used to the whole query(no specific
- * nodes).
- */
- public static final HintPredicate SET_VAR =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.SET_VAR);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Join} nodes.
- */
- public static final HintPredicate JOIN =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.JOIN);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.TableScan} nodes.
- */
- public static final HintPredicate TABLE_SCAN =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.TABLE_SCAN);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Filter} nodes.
- */
- public static final HintPredicate FILTER =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.FILTER);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Project} nodes.
- */
- public static final HintPredicate PROJECT =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.PROJECT);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Correlate} nodes.
- */
- public static final HintPredicate CORRELATE =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.CORRELATE);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.SetOp} nodes.
- */
- public static final HintPredicate SETOP =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.SETOP);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Sort} nodes.
- */
- public static final HintPredicate SORT =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.SORT);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Aggregate} nodes.
- */
- public static final HintPredicate AGGREGATE =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.AGGREGATE);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Calc} nodes.
- */
- public static final HintPredicate CALC =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.CALC);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Values} nodes.
- */
- public static final HintPredicate VALUES =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.VALUES);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Window} nodes.
- */
- public static final HintPredicate WINDOW =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.WINDOW);
-
- /**
- * A hint predicate that indicates a hint can only be used to {@link
- * org.apache.calcite.rel.core.Snapshot} nodes.
- */
- public static final HintPredicate SNAPSHOT =
- new NodeTypeHintPredicate(NodeTypeHintPredicate.NodeType.SNAPSHOT);
-
- /**
- * Returns a composed hint predicate that represents a short-circuiting logical AND of an array
- * of hint predicates {@code hintPredicates}. When evaluating the composed predicate, if a
- * predicate is {@code false}, then all the left predicates are not evaluated.
- *
- * <p>The predicates are evaluated in sequence.
- */
- public static HintPredicate and(HintPredicate... hintPredicates) {
- return new CompositeHintPredicate(CompositeHintPredicate.Composition.AND, hintPredicates);
- }
-
- /**
- * Returns a composed hint predicate that represents a short-circuiting logical OR of an array
- * of hint predicates {@code hintPredicates}. When evaluating the composed predicate, if a
- * predicate is {@code true}, then all the left predicates are not evaluated.
- *
- * <p>The predicates are evaluated in sequence.
- */
- public static HintPredicate or(HintPredicate... hintPredicates) {
- return new CompositeHintPredicate(CompositeHintPredicate.Composition.OR, hintPredicates);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalFilter.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalFilter.java
deleted file mode 100644
index 6aecf9ceab7..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalFilter.java
+++ /dev/null
@@ -1,165 +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 com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
-import org.apache.calcite.plan.Convention;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollationTraitDef;
-import org.apache.calcite.rel.RelDistributionTraitDef;
-import org.apache.calcite.rel.RelInput;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelShuttle;
-import org.apache.calcite.rel.RelWriter;
-import org.apache.calcite.rel.core.CorrelationId;
-import org.apache.calcite.rel.core.Filter;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rel.metadata.RelMdCollation;
-import org.apache.calcite.rel.metadata.RelMdDistribution;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.rex.RexNode;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * Sub-class of {@link org.apache.calcite.rel.core.Filter} not targeted at any particular engine or
- * calling convention.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public final class LogicalFilter extends Filter {
- private final ImmutableSet<CorrelationId> variablesSet;
-
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a LogicalFilter.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param hints Hints for this node
- * @param child Input relational expression
- * @param condition Boolean expression which determines whether a row is allowed to pass
- * @param variablesSet Correlation variables set by this relational expression to be used by
- * nested expressions
- */
- public LogicalFilter(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- RelNode child,
- RexNode condition,
- ImmutableSet<CorrelationId> variablesSet) {
- super(cluster, traitSet, hints, child, condition);
- this.variablesSet = Objects.requireNonNull(variablesSet, "variablesSet");
- }
-
- @Deprecated // to be removed before 2.0
- public LogicalFilter(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelNode child,
- RexNode condition,
- com.google.common.collect.ImmutableSet<CorrelationId> variablesSet) {
- this(cluster, traitSet, ImmutableList.of(), child, condition, variablesSet);
- }
-
- @Deprecated // to be removed before 2.0
- public LogicalFilter(
- RelOptCluster cluster, RelTraitSet traitSet, RelNode child, RexNode condition) {
- this(cluster, traitSet, child, condition, ImmutableSet.of());
- }
-
- @Deprecated // to be removed before 2.0
- public LogicalFilter(RelOptCluster cluster, RelNode child, RexNode condition) {
- this(cluster, cluster.traitSetOf(Convention.NONE), child, condition, ImmutableSet.of());
- }
-
- /** Creates a LogicalFilter by parsing serialized output. */
- public LogicalFilter(RelInput input) {
- super(input);
- this.variablesSet = ImmutableSet.of();
- }
-
- /** Creates a LogicalFilter. */
- public static LogicalFilter create(final RelNode input, RexNode condition) {
- return create(input, condition, ImmutableSet.of());
- }
-
- /** Creates a LogicalFilter. */
- public static LogicalFilter create(
- final RelNode input, RexNode condition, ImmutableSet<CorrelationId> variablesSet) {
- final RelOptCluster cluster = input.getCluster();
- final RelMetadataQuery mq = cluster.getMetadataQuery();
- final RelTraitSet traitSet =
- cluster.traitSetOf(Convention.NONE)
- .replaceIfs(
- RelCollationTraitDef.INSTANCE,
- () -> RelMdCollation.filter(mq, input))
- .replaceIf(
- RelDistributionTraitDef.INSTANCE,
- () -> RelMdDistribution.filter(mq, input));
- return new LogicalFilter(cluster, traitSet, input, condition, variablesSet);
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public Set<CorrelationId> getVariablesSet() {
- return variablesSet;
- }
-
- @Override
- public LogicalFilter copy(RelTraitSet traitSet, RelNode input, RexNode condition) {
- assert traitSet.containsIfApplicable(Convention.NONE);
- return new LogicalFilter(getCluster(), traitSet, hints, input, condition, variablesSet);
- }
-
- @Override
- public RelNode accept(RelShuttle shuttle) {
- return shuttle.visit(this);
- }
-
- @Override
- public RelWriter explainTerms(RelWriter pw) {
- return super.explainTerms(pw).itemIf("variablesSet", variablesSet, !variablesSet.isEmpty());
- }
-
- @Override
- public boolean deepEquals(@Nullable Object obj) {
- return deepEquals0(obj) && variablesSet.equals(((LogicalFilter) obj).variablesSet);
- }
-
- @Override
- public int deepHashCode() {
- return Objects.hash(deepHashCode0(), variablesSet);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalFilter(getCluster(), traitSet, hintList, input, condition, variablesSet);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalIntersect.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalIntersect.java
deleted file mode 100644
index b3975937604..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalIntersect.java
+++ /dev/null
@@ -1,97 +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.Intersect;
-import org.apache.calcite.rel.hint.RelHint;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Sub-class of {@link org.apache.calcite.rel.core.Intersect} not targeted at any particular engine
- * or calling convention.
- */
-public final class LogicalIntersect extends Intersect {
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a LogicalIntersect.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- */
- public LogicalIntersect(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- List<RelNode> inputs,
- boolean all) {
- super(cluster, traitSet, hints, inputs, all);
- }
-
- /**
- * Creates a LogicalIntersect.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- */
- public LogicalIntersect(
- RelOptCluster cluster, RelTraitSet traitSet, List<RelNode> inputs, boolean all) {
- this(cluster, traitSet, Collections.emptyList(), inputs, all);
- }
-
- @Deprecated // to be removed before 2.0
- public LogicalIntersect(RelOptCluster cluster, List<RelNode> inputs, boolean all) {
- this(cluster, cluster.traitSetOf(Convention.NONE), inputs, all);
- }
-
- /** Creates a LogicalIntersect by parsing serialized output. */
- public LogicalIntersect(RelInput input) {
- super(input);
- }
-
- /** Creates a LogicalIntersect. */
- public static LogicalIntersect create(List<RelNode> inputs, boolean all) {
- final RelOptCluster cluster = inputs.get(0).getCluster();
- final RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE);
- return new LogicalIntersect(cluster, traitSet, inputs, all);
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public LogicalIntersect copy(RelTraitSet traitSet, List<RelNode> inputs, boolean all) {
- return new LogicalIntersect(getCluster(), traitSet, hints, inputs, all);
- }
-
- @Override
- public RelNode accept(RelShuttle shuttle) {
- return shuttle.visit(this);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalIntersect(getCluster(), traitSet, hintList, inputs, all);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalMinus.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalMinus.java
deleted file mode 100644
index cf04c1720da..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalMinus.java
+++ /dev/null
@@ -1,101 +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.Minus;
-import org.apache.calcite.rel.hint.RelHint;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Sub-class of {@link org.apache.calcite.rel.core.Minus} not targeted at any particular engine or
- * calling convention.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public final class LogicalMinus extends Minus {
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a LogicalMinus.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- */
- public LogicalMinus(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- List<RelNode> inputs,
- boolean all) {
- super(cluster, traitSet, hints, inputs, all);
- }
-
- /**
- * Creates a LogicalMinus.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- */
- public LogicalMinus(
- RelOptCluster cluster, RelTraitSet traitSet, List<RelNode> inputs, boolean all) {
- this(cluster, traitSet, Collections.emptyList(), inputs, all);
- }
-
- @Deprecated // to be removed before 2.0
- public LogicalMinus(RelOptCluster cluster, List<RelNode> inputs, boolean all) {
- this(cluster, cluster.traitSetOf(Convention.NONE), inputs, all);
- }
-
- /** Creates a LogicalMinus by parsing serialized output. */
- public LogicalMinus(RelInput input) {
- super(input);
- }
-
- /** Creates a LogicalMinus. */
- public static LogicalMinus create(List<RelNode> inputs, boolean all) {
- final RelOptCluster cluster = inputs.get(0).getCluster();
- final RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE);
- return new LogicalMinus(cluster, traitSet, inputs, all);
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public LogicalMinus copy(RelTraitSet traitSet, List<RelNode> inputs, boolean all) {
- assert traitSet.containsIfApplicable(Convention.NONE);
- return new LogicalMinus(getCluster(), traitSet, hints, inputs, all);
- }
-
- @Override
- public RelNode accept(RelShuttle shuttle) {
- return shuttle.visit(this);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalMinus(getCluster(), traitSet, hintList, inputs, all);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalSort.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalSort.java
deleted file mode 100644
index 66e037b4523..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalSort.java
+++ /dev/null
@@ -1,113 +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.RelCollation;
-import org.apache.calcite.rel.RelCollationTraitDef;
-import org.apache.calcite.rel.RelInput;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelShuttle;
-import org.apache.calcite.rel.core.Sort;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rex.RexNode;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Sub-class of {@link org.apache.calcite.rel.core.Sort} not targeted at any particular engine or
- * calling convention.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public final class LogicalSort extends Sort {
- private LogicalSort(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelNode input,
- RelCollation collation,
- RexNode offset,
- RexNode fetch) {
- this(cluster, traitSet, Collections.emptyList(), input, collation, offset, fetch);
- }
-
- private LogicalSort(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- RelNode input,
- RelCollation collation,
- @Nullable RexNode offset,
- @Nullable RexNode fetch) {
- super(cluster, traitSet, hints, input, collation, offset, fetch);
- assert traitSet.containsIfApplicable(Convention.NONE);
- }
-
- /** Creates a LogicalSort by parsing serialized output. */
- public LogicalSort(RelInput input) {
- super(input);
- }
-
- /**
- * Creates a LogicalSort.
- *
- * @param input Input relational expression
- * @param collation array of sort specifications
- * @param offset Expression for number of rows to discard before returning first row
- * @param fetch Expression for number of rows to fetch
- */
- public static LogicalSort create(
- RelNode input,
- RelCollation collation,
- @Nullable RexNode offset,
- @Nullable RexNode fetch) {
- RelOptCluster cluster = input.getCluster();
- collation = RelCollationTraitDef.INSTANCE.canonize(collation);
- RelTraitSet traitSet = input.getTraitSet().replace(Convention.NONE).replace(collation);
- return new LogicalSort(cluster, traitSet, input, collation, offset, fetch);
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public Sort copy(
- RelTraitSet traitSet,
- RelNode newInput,
- RelCollation newCollation,
- @Nullable RexNode offset,
- @Nullable RexNode fetch) {
- return new LogicalSort(
- getCluster(), traitSet, hints, newInput, newCollation, offset, fetch);
- }
-
- @Override
- public RelNode accept(RelShuttle shuttle) {
- return shuttle.visit(this);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalSort(getCluster(), traitSet, hintList, input, collation, offset, fetch);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalUnion.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalUnion.java
deleted file mode 100644
index 2e1992f6f5a..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalUnion.java
+++ /dev/null
@@ -1,101 +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.Union;
-import org.apache.calcite.rel.hint.RelHint;
-
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Sub-class of {@link org.apache.calcite.rel.core.Union} not targeted at any particular engine or
- * calling convention.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public final class LogicalUnion extends Union {
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a LogicalUnion.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- */
- public LogicalUnion(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- List<RelNode> inputs,
- boolean all) {
- super(cluster, traitSet, hints, inputs, all);
- }
-
- /**
- * Creates a LogicalUnion.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- */
- public LogicalUnion(
- RelOptCluster cluster, RelTraitSet traitSet, List<RelNode> inputs, boolean all) {
- this(cluster, traitSet, Collections.emptyList(), inputs, all);
- }
-
- @Deprecated // to be removed before 2.0
- public LogicalUnion(RelOptCluster cluster, List<RelNode> inputs, boolean all) {
- this(cluster, cluster.traitSetOf(Convention.NONE), inputs, all);
- }
-
- /** Creates a LogicalUnion by parsing serialized output. */
- public LogicalUnion(RelInput input) {
- super(input);
- }
-
- /** Creates a LogicalUnion. */
- public static LogicalUnion create(List<RelNode> inputs, boolean all) {
- final RelOptCluster cluster = inputs.get(0).getCluster();
- final RelTraitSet traitSet = cluster.traitSetOf(Convention.NONE);
- return new LogicalUnion(cluster, traitSet, inputs, all);
- }
-
- // ~ Methods ----------------------------------------------------------------
-
- @Override
- public LogicalUnion copy(RelTraitSet traitSet, List<RelNode> inputs, boolean all) {
- assert traitSet.containsIfApplicable(Convention.NONE);
- return new LogicalUnion(getCluster(), traitSet, hints, inputs, all);
- }
-
- @Override
- public RelNode accept(RelShuttle shuttle) {
- return shuttle.visit(this);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalUnion(getCluster(), traitSet, hintList, inputs, all);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalValues.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalValues.java
deleted file mode 100644
index e76a9ecf192..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalValues.java
+++ /dev/null
@@ -1,160 +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 com.google.common.collect.ImmutableList;
-import org.apache.calcite.plan.Convention;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollationTraitDef;
-import org.apache.calcite.rel.RelDistributionTraitDef;
-import org.apache.calcite.rel.RelInput;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelShuttle;
-import org.apache.calcite.rel.core.Values;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rel.metadata.RelMdCollation;
-import org.apache.calcite.rel.metadata.RelMdDistribution;
-import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.sql.type.SqlTypeName;
-
-import java.math.BigDecimal;
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Sub-class of {@link org.apache.calcite.rel.core.Values} not targeted at any particular engine or
- * calling convention.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public class LogicalValues extends Values {
- // ~ Constructors -----------------------------------------------------------
-
- /**
- * Creates a LogicalValues.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param hints Hints for this node
- * @param rowType Row type for tuples produced by this rel
- * @param tuples 2-dimensional array of tuple values to be produced; outer list contains tuples;
- * each inner list is one tuple; all tuples must be of same length, conforming to rowType
- */
- public LogicalValues(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- RelDataType rowType,
- ImmutableList<ImmutableList<RexLiteral>> tuples) {
- super(cluster, hints, rowType, tuples, traitSet);
- }
-
- /**
- * Creates a LogicalValues.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param rowType Row type for tuples produced by this rel
- * @param tuples 2-dimensional array of tuple values to be produced; outer list contains tuples;
- * each inner list is one tuple; all tuples must be of same length, conforming to rowType
- */
- public LogicalValues(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelDataType rowType,
- ImmutableList<ImmutableList<RexLiteral>> tuples) {
- this(cluster, traitSet, Collections.emptyList(), rowType, tuples);
- }
-
- @Deprecated // to be removed before 2.0
- public LogicalValues(
- RelOptCluster cluster,
- RelDataType rowType,
- ImmutableList<ImmutableList<RexLiteral>> tuples) {
- this(cluster, cluster.traitSetOf(Convention.NONE), rowType, tuples);
- }
-
- /** Creates a LogicalValues by parsing serialized output. */
- public LogicalValues(RelInput input) {
- super(input);
- }
-
- /** Creates a LogicalValues. */
- public static LogicalValues create(
- RelOptCluster cluster,
- final RelDataType rowType,
- final ImmutableList<ImmutableList<RexLiteral>> tuples) {
- final RelMetadataQuery mq = cluster.getMetadataQuery();
- final RelTraitSet traitSet =
- cluster.traitSetOf(Convention.NONE)
- .replaceIfs(
- RelCollationTraitDef.INSTANCE,
- () -> RelMdCollation.values(mq, rowType, tuples))
- .replaceIf(
- RelDistributionTraitDef.INSTANCE,
- () -> RelMdDistribution.values(rowType, tuples));
- return new LogicalValues(cluster, traitSet, rowType, tuples);
- }
-
- @Override
- public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
- assert traitSet.containsIfApplicable(Convention.NONE);
- assert inputs.isEmpty();
- return new LogicalValues(getCluster(), traitSet, rowType, tuples);
- }
-
- /** Creates a LogicalValues that outputs no rows of a given row type. */
- public static LogicalValues createEmpty(RelOptCluster cluster, RelDataType rowType) {
- return create(cluster, rowType, ImmutableList.of());
- }
-
- /** Creates a LogicalValues that outputs one row and one column. */
- public static LogicalValues createOneRow(RelOptCluster cluster) {
- final RelDataType rowType =
- cluster.getTypeFactory()
- .builder()
- .add("ZERO", SqlTypeName.INTEGER)
- .nullable(false)
- .build();
- final ImmutableList<ImmutableList<RexLiteral>> tuples =
- ImmutableList.of(
- ImmutableList.of(
- cluster.getRexBuilder()
- .makeExactLiteral(
- BigDecimal.ZERO,
- rowType.getFieldList().get(0).getType())));
- return create(cluster, rowType, tuples);
- }
-
- @Override
- public RelNode accept(RelShuttle shuttle) {
- return shuttle.visit(this);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalValues(getCluster(), traitSet, hintList, rowType, tuples);
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java
deleted file mode 100644
index 3208f4205d8..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/logical/LogicalWindow.java
+++ /dev/null
@@ -1,404 +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 com.google.common.collect.LinkedListMultimap;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Multimap;
-import org.apache.calcite.linq4j.Ord;
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptUtil;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollation;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.core.Window;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rex.RexInputRef;
-import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.rex.RexLocalRef;
-import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.rex.RexOver;
-import org.apache.calcite.rex.RexProgram;
-import org.apache.calcite.rex.RexShuttle;
-import org.apache.calcite.rex.RexWindow;
-import org.apache.calcite.rex.RexWindowBound;
-import org.apache.calcite.tools.RelBuilder;
-import org.apache.calcite.util.ImmutableBitSet;
-import org.apache.calcite.util.Litmus;
-import org.apache.calcite.util.Pair;
-import org.apache.calcite.util.Util;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.AbstractList;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.IdentityHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-
-/**
- * Sub-class of {@link org.apache.calcite.rel.core.Window} not targeted at any particular engine or
- * calling convention.
- *
- * <p>Temporarily copy from calcite to cherry-pick [CALCITE-5107] and will be removed when upgrade
- * the latest calcite.
- */
-public final class LogicalWindow extends Window {
- /**
- * Creates a LogicalWindow.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- *
- * @param cluster Cluster
- * @param traitSet Trait set
- * @param hints hints for this node
- * @param input Input relational expression
- * @param constants List of constants that are additional inputs
- * @param rowType Output row type
- * @param groups Window groups
- */
- public LogicalWindow(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- List<RelHint> hints,
- RelNode input,
- List<RexLiteral> constants,
- RelDataType rowType,
- List<Group> groups) {
- super(cluster, traitSet, hints, input, constants, rowType, groups);
- }
-
- /**
- * Creates a LogicalWindow.
- *
- * <p>Use {@link #create} unless you know what you're doing.
- */
- public LogicalWindow(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelNode input,
- List<RexLiteral> constants,
- RelDataType rowType,
- List<Group> groups) {
- this(cluster, traitSet, Collections.emptyList(), input, constants, rowType, groups);
- }
-
- @Override
- public LogicalWindow copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return new LogicalWindow(
- getCluster(), traitSet, hints, sole(inputs), constants, rowType, groups);
- }
-
- /**
- * Creates a LogicalWindow.
- *
- * @param input Input relational expression
- * @param traitSet Trait set
- * @param constants List of constants that are additional inputs
- * @param rowType Output row type
- * @param groups Window groups
- */
- public static LogicalWindow create(
- RelTraitSet traitSet,
- RelNode input,
- List<RexLiteral> constants,
- RelDataType rowType,
- List<Window.Group> groups) {
- return new LogicalWindow(input.getCluster(), traitSet, input, constants, rowType, groups);
- }
-
- /** Creates a LogicalWindow by parsing a {@link RexProgram}. */
- public static RelNode create(
- RelOptCluster cluster,
- RelTraitSet traitSet,
- RelBuilder relBuilder,
- RelNode child,
- final RexProgram program) {
- final RelDataType outRowType = program.getOutputRowType();
- // Build a list of distinct groups, partitions and aggregate
- // functions.
- final Multimap<WindowKey, RexOver> windowMap = LinkedListMultimap.create();
-
- final int inputFieldCount = child.getRowType().getFieldCount();
-
- final Map<RexLiteral, RexInputRef> constantPool = new HashMap<>();
- final List<RexLiteral> constants = new ArrayList<>();
-
- // Identify constants in the expression tree and replace them with
- // references to newly generated constant pool.
- RexShuttle replaceConstants =
- new RexShuttle() {
- @Override
- public RexNode visitLiteral(RexLiteral literal) {
- RexInputRef ref = constantPool.get(literal);
- if (ref != null) {
- return ref;
- }
- constants.add(literal);
- ref =
- new RexInputRef(
- constantPool.size() + inputFieldCount, literal.getType());
- constantPool.put(literal, ref);
- return ref;
- }
- };
-
- // Build a list of groups, partitions, and aggregate functions. Each
- // aggregate function will add its arguments as outputs of the input
- // program.
- final IdentityHashMap<RexOver, RexOver> origToNewOver = new IdentityHashMap<>();
- for (RexNode agg : program.getExprList()) {
- if (agg instanceof RexOver) {
- final RexOver origOver = (RexOver) agg;
- final RexOver newOver = (RexOver) origOver.accept(replaceConstants);
- origToNewOver.put(origOver, newOver);
- addWindows(windowMap, newOver, inputFieldCount);
- }
- }
-
- final Map<RexOver, Window.RexWinAggCall> aggMap = new HashMap<>();
- List<Group> groups = new ArrayList<>();
- for (Map.Entry<WindowKey, Collection<RexOver>> entry : windowMap.asMap().entrySet()) {
- final WindowKey windowKey = entry.getKey();
- final List<RexWinAggCall> aggCalls = new ArrayList<>();
- for (RexOver over : entry.getValue()) {
- final RexWinAggCall aggCall =
- new RexWinAggCall(
- over.getAggOperator(),
- over.getType(),
- toInputRefs(over.operands),
- aggMap.size(),
- over.isDistinct(),
- over.ignoreNulls());
- aggCalls.add(aggCall);
- aggMap.put(over, aggCall);
- }
- RexShuttle toInputRefs =
- new RexShuttle() {
- @Override
- public RexNode visitLocalRef(RexLocalRef localRef) {
- return new RexInputRef(localRef.getIndex(), localRef.getType());
- }
- };
- groups.add(
- new Group(
- windowKey.groupSet,
- windowKey.isRows,
- windowKey.lowerBound.accept(toInputRefs),
- windowKey.upperBound.accept(toInputRefs),
- windowKey.orderKeys,
- aggCalls));
- }
-
- // Figure out the type of the inputs to the output program.
- // They are: the inputs to this rel, followed by the outputs of
- // each window.
- final List<Window.RexWinAggCall> flattenedAggCallList = new ArrayList<>();
- final List<Map.Entry<String, RelDataType>> fieldList =
- new ArrayList<>(child.getRowType().getFieldList());
- final int offset = fieldList.size();
-
- // Use better field names for agg calls that are projected.
- final Map<Integer, String> fieldNames = new HashMap<>();
- for (Ord<RexLocalRef> ref : Ord.zip(program.getProjectList())) {
- final int index = ref.e.getIndex();
- if (index >= offset) {
- fieldNames.put(index - offset, outRowType.getFieldNames().get(ref.i));
- }
- }
-
- for (Ord<Group> window : Ord.zip(groups)) {
- for (Ord<RexWinAggCall> over : Ord.zip(window.e.aggCalls)) {
- // Add the k-th over expression of
- // the i-th window to the output of the program.
- String name = fieldNames.get(over.i);
- if (name == null || name.startsWith("$")) {
- name = "w" + window.i + "$o" + over.i;
- }
- fieldList.add(Pair.of(name, over.e.getType()));
- flattenedAggCallList.add(over.e);
- }
- }
- final RelDataType intermediateRowType =
- cluster.getTypeFactory().createStructType(fieldList);
-
- // The output program is the windowed agg's program, combined with
- // the output calc (if it exists).
- RexShuttle shuttle =
- new RexShuttle() {
- @Override
- public RexNode visitOver(RexOver over) {
- // Look up the aggCall which this expr was translated to.
- final Window.RexWinAggCall aggCall = aggMap.get(origToNewOver.get(over));
- assert aggCall != null;
- assert RelOptUtil.eq(
- "over", over.getType(), "aggCall", aggCall.getType(), Litmus.THROW);
-
- // Find the index of the aggCall among all partitions of all
- // groups.
- final int aggCallIndex = flattenedAggCallList.indexOf(aggCall);
- assert aggCallIndex >= 0;
-
- // Replace expression with a reference to the window slot.
- final int index = inputFieldCount + aggCallIndex;
- assert RelOptUtil.eq(
- "over",
- over.getType(),
- "intermed",
- intermediateRowType.getFieldList().get(index).getType(),
- Litmus.THROW);
- return new RexInputRef(index, over.getType());
- }
-
- @Override
- public RexNode visitLocalRef(RexLocalRef localRef) {
- final int index = localRef.getIndex();
- if (index < inputFieldCount) {
- // Reference to input field.
- return localRef;
- }
- return new RexLocalRef(
- flattenedAggCallList.size() + index, localRef.getType());
- }
- };
-
- final LogicalWindow window =
- LogicalWindow.create(traitSet, child, constants, intermediateRowType, groups);
-
- // The order that the "over" calls occur in the groups and
- // partitions may not match the order in which they occurred in the
- // original expression.
- // Add a project to permute them.
- final List<RexNode> refToWindow = toInputRefs(shuttle.visitList(program.getExprList()));
-
- final List<RexNode> projectList = new ArrayList<>();
- for (RexLocalRef inputRef : program.getProjectList()) {
- final int index = inputRef.getIndex();
- final RexInputRef ref = (RexInputRef) refToWindow.get(index);
- projectList.add(ref);
- }
-
- return relBuilder.push(window).project(projectList, outRowType.getFieldNames()).build();
- }
-
- private static List<RexNode> toInputRefs(final List<? extends RexNode> operands) {
- return new AbstractList<RexNode>() {
- @Override
- public int size() {
- return operands.size();
- }
-
- @Override
- public RexNode get(int index) {
- final RexNode operand = operands.get(index);
- if (operand instanceof RexInputRef) {
- return operand;
- }
- assert operand instanceof RexLocalRef;
- final RexLocalRef ref = (RexLocalRef) operand;
- return new RexInputRef(ref.getIndex(), ref.getType());
- }
- };
- }
-
- /**
- * Group specification. All windowed aggregates over the same window (regardless of how it is
- * specified, in terms of a named window or specified attribute by attribute) will end up with
- * the same window key.
- */
- private static class WindowKey {
- private final ImmutableBitSet groupSet;
- private final RelCollation orderKeys;
- private final boolean isRows;
- private final RexWindowBound lowerBound;
- private final RexWindowBound upperBound;
-
- WindowKey(
- ImmutableBitSet groupSet,
- RelCollation orderKeys,
- boolean isRows,
- RexWindowBound lowerBound,
- RexWindowBound upperBound) {
- this.groupSet = groupSet;
- this.orderKeys = orderKeys;
- this.isRows = isRows;
- this.lowerBound = lowerBound;
- this.upperBound = upperBound;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(groupSet, orderKeys, isRows, lowerBound, upperBound);
- }
-
- @Override
- public boolean equals(@Nullable Object obj) {
- return obj == this
- || obj instanceof WindowKey
- && groupSet.equals(((WindowKey) obj).groupSet)
- && orderKeys.equals(((WindowKey) obj).orderKeys)
- && Objects.equals(lowerBound, ((WindowKey) obj).lowerBound)
- && Objects.equals(upperBound, ((WindowKey) obj).upperBound)
- && isRows == ((WindowKey) obj).isRows;
- }
- }
-
- private static void addWindows(
- Multimap<WindowKey, RexOver> windowMap, RexOver over, final int inputFieldCount) {
- final RexWindow aggWindow = over.getWindow();
-
- // Look up or create a window.
- RelCollation orderKeys =
- getCollation(
- Lists.newArrayList(
- Util.filter(
- aggWindow.orderKeys,
- rexFieldCollation ->
- // If ORDER BY references constant (i.e.
- // RexInputRef),
- // then we can ignore such ORDER BY key.
- rexFieldCollation.left instanceof RexLocalRef)));
- ImmutableBitSet groupSet = ImmutableBitSet.of(getProjectOrdinals(aggWindow.partitionKeys));
- final int groupLength = groupSet.length();
- if (inputFieldCount < groupLength) {
- // If PARTITION BY references constant, we can ignore such partition key.
- // All the inputs after inputFieldCount are literals, thus we can clear.
- groupSet = groupSet.except(ImmutableBitSet.range(inputFieldCount, groupLength));
- }
-
- WindowKey windowKey =
- new WindowKey(
- groupSet,
- orderKeys,
- aggWindow.isRows(),
- aggWindow.getLowerBound(),
- aggWindow.getUpperBound());
- windowMap.put(windowKey, over);
- }
-
- @Override
- public RelNode withHints(List<RelHint> hintList) {
- return new LogicalWindow(
- getCluster(), traitSet, hintList, input, constants, rowType, groups);
- }
-}
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 d05271f03c1..78808db17e1 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
@@ -85,6 +85,7 @@ import org.apache.calcite.sql.SqlWindow;
import org.apache.calcite.sql.SqlWindowTableFunction;
import org.apache.calcite.sql.SqlWith;
import org.apache.calcite.sql.SqlWithItem;
+import org.apache.calcite.sql.TableCharacteristic;
import org.apache.calcite.sql.fun.SqlCase;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.parser.SqlParserPos;
@@ -157,10 +158,10 @@ import static org.apache.calcite.util.Static.RESOURCE;
* Default implementation of {@link SqlValidator}, the class was copied over because of
* CALCITE-4554.
*
- * <p>Lines 4995 ~ 5008, Flink enables TIMESTAMP and TIMESTAMP_LTZ for system time period
+ * <p>Lines 5083 ~ 5096, Flink enables TIMESTAMP and TIMESTAMP_LTZ for system time period
* specification type at {@link org.apache.calcite.sql.validate.SqlValidatorImpl#validateSnapshot}.
*
- * <p>Lines 5352 ~ 5358, Flink enables TIMESTAMP and TIMESTAMP_LTZ for first orderBy column in
+ * <p>Lines 5440 ~ 5446, Flink enables TIMESTAMP and TIMESTAMP_LTZ for first orderBy column in
* matchRecognize at {@link SqlValidatorImpl#validateMatchRecognize}.
*/
public class SqlValidatorImpl implements SqlValidatorWithHints {
@@ -446,7 +447,14 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
if (expanded != null) {
inferUnknownTypes(targetType, scope, expanded);
}
- final RelDataType type = deriveType(selectScope, expanded);
+ RelDataType type = deriveType(selectScope, expanded);
+ // Re-derive SELECT ITEM's data type that may be nullable in AggregatingSelectScope when it
+ // appears in advanced grouping elements such as CUBE, ROLLUP , GROUPING SETS.
+ // For example, SELECT CASE WHEN c = 1 THEN '1' ELSE '23' END AS x FROM t GROUP BY CUBE(x),
+ // the 'x' should be nullable even if x's literal values are not null.
+ if (selectScope instanceof AggregatingSelectScope) {
+ type = requireNonNull(selectScope.nullifyType(stripAs(expanded), type));
+ }
setValidatedNodeType(expanded, type);
fields.add(Pair.of(alias, type));
return false;
@@ -501,26 +509,26 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
public @Nullable List<String> usingNames(SqlJoin join) {
switch (join.getConditionType()) {
case USING:
- final ImmutableList.Builder<String> list = ImmutableList.builder();
- final Set<String> names = catalogReader.nameMatcher().createSet();
- for (String name : SqlIdentifier.simpleNames((SqlNodeList) getCondition(join))) {
- if (names.add(name)) {
- list.add(name);
- }
- }
- return list.build();
+ SqlNodeList condition = (SqlNodeList) getCondition(join);
+ List<String> simpleNames = SqlIdentifier.simpleNames(condition);
+ return catalogReader.nameMatcher().distinctCopy(simpleNames);
+
case NONE:
if (join.isNatural()) {
- final RelDataType t0 = getValidatedNodeType(join.getLeft());
- final RelDataType t1 = getValidatedNodeType(join.getRight());
- return SqlValidatorUtil.deriveNaturalJoinColumnList(
- catalogReader.nameMatcher(), t0, t1);
+ return deriveNaturalJoinColumnList(join);
}
- break;
+ return null;
+
default:
- break;
+ return null;
}
- return null;
+ }
+
+ private List<String> deriveNaturalJoinColumnList(SqlJoin join) {
+ return SqlValidatorUtil.deriveNaturalJoinColumnList(
+ catalogReader.nameMatcher(),
+ getNamespaceOrThrow(join.getLeft()).getRowType(),
+ getNamespaceOrThrow(join.getRight()).getRowType());
}
private static SqlNode expandCommonColumn(
@@ -3283,6 +3291,9 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
case UNNEST:
validateUnnest((SqlCall) node, scope, targetRowType);
break;
+ case COLLECTION_TABLE:
+ validateTableFunction((SqlCall) node, scope, targetRowType);
+ break;
default:
validateQuery(node, scope, targetRowType);
break;
@@ -3293,6 +3304,65 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
getNamespaceOrThrow(node, scope).validate(targetRowType);
}
+ protected void validateTableFunction(
+ SqlCall node, SqlValidatorScope scope, RelDataType targetRowType) {
+ // Dig out real call; TABLE() wrapper is just syntactic.
+ SqlCall call = node.operand(0);
+ if (call.getOperator() instanceof SqlTableFunction) {
+ SqlTableFunction tableFunction = (SqlTableFunction) call.getOperator();
+ boolean visitedRowSemanticsTable = false;
+ for (int idx = 0; idx < call.operandCount(); idx++) {
+ TableCharacteristic tableCharacteristic = tableFunction.tableCharacteristic(idx);
+ if (tableCharacteristic != null) {
+ // Skip validate if current input table has set semantics
+ if (tableCharacteristic.semantics == TableCharacteristic.Semantics.SET) {
+ continue;
+ }
+ // A table function at most has one input table with row semantics
+ if (visitedRowSemanticsTable) {
+ throw newValidationError(
+ call,
+ RESOURCE.multipleRowSemanticsTables(call.getOperator().getName()));
+ }
+ visitedRowSemanticsTable = true;
+ }
+ // If table function defines the parameter is not table parameter, or is an input
+ // table
+ // parameter with row semantics, then it should not be with PARTITION BY OR ORDER
+ // BY.
+ SqlNode currentNode = call.operand(idx);
+ if (currentNode instanceof SqlCall) {
+ SqlOperator op = ((SqlCall) currentNode).getOperator();
+ if (op == SqlStdOperatorTable.ARGUMENT_ASSIGNMENT) {
+ // Dig out the underlying operand
+ SqlNode realNode = ((SqlBasicCall) currentNode).operand(0);
+ if (realNode instanceof SqlCall) {
+ currentNode = realNode;
+ op = ((SqlCall) realNode).getOperator();
+ }
+ }
+ if (op == SqlStdOperatorTable.SET_SEMANTICS_TABLE) {
+ throwInvalidRowSemanticsTable(call, idx, (SqlCall) currentNode);
+ }
+ }
+ }
+ }
+ validateQuery(node, scope, targetRowType);
+ }
+
+ private void throwInvalidRowSemanticsTable(SqlCall call, int idx, SqlCall table) {
+ SqlNodeList partitionList = table.operand(1);
+ if (!partitionList.isEmpty()) {
+ throw newValidationError(
+ call, RESOURCE.invalidPartitionKeys(idx, call.getOperator().getName()));
+ }
+ SqlNodeList orderList = table.operand(2);
+ if (!orderList.isEmpty()) {
+ throw newValidationError(
+ call, RESOURCE.invalidOrderBy(idx, call.getOperator().getName()));
+ }
+ }
+
protected void validateOver(SqlCall call, SqlValidatorScope scope) {
throw new AssertionError("OVER unexpected in this context");
}
@@ -3324,10 +3394,9 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
}
protected void validateJoin(SqlJoin join, SqlValidatorScope scope) {
- SqlNode left = join.getLeft();
- SqlNode right = join.getRight();
- SqlNode condition = join.getCondition();
- boolean natural = join.isNatural();
+ final SqlNode left = join.getLeft();
+ final SqlNode right = join.getRight();
+ final boolean natural = join.isNatural();
final JoinType joinType = join.getJoinType();
final JoinConditionType conditionType = join.getConditionType();
final SqlValidatorScope joinScope = getScopeOrThrow(join); // getJoinScope?
@@ -3337,35 +3406,22 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
// Validate condition.
switch (conditionType) {
case NONE:
- Preconditions.checkArgument(condition == null);
+ Preconditions.checkArgument(join.getCondition() == null);
break;
case ON:
- requireNonNull(condition, "join.getCondition()");
- SqlNode expandedCondition = expand(condition, joinScope);
- join.setOperand(5, expandedCondition);
- condition = getCondition(join);
+ final SqlNode condition = expand(getCondition(join), joinScope);
+ join.setOperand(5, condition);
validateWhereOrOn(joinScope, condition, "ON");
checkRollUp(null, join, condition, joinScope, "ON");
break;
case USING:
- SqlNodeList list = (SqlNodeList) requireNonNull(condition, "join.getCondition()");
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ List<SqlIdentifier> list = (List) getCondition(join);
// Parser ensures that using clause is not empty.
Preconditions.checkArgument(list.size() > 0, "Empty USING clause");
- for (SqlNode node : list) {
- SqlIdentifier id = (SqlIdentifier) node;
- final RelDataType leftColType = validateUsingCol(id, left);
- final RelDataType rightColType = validateUsingCol(id, right);
- if (!SqlTypeUtil.isComparable(leftColType, rightColType)) {
- throw newValidationError(
- id,
- RESOURCE.naturalOrUsingColumnNotCompatible(
- id.getSimple(),
- leftColType.toString(),
- rightColType.toString()));
- }
- checkRollUpInUsing(id, left, scope);
- checkRollUpInUsing(id, right, scope);
+ for (SqlIdentifier id : list) {
+ validateCommonJoinColumn(id, left, right, scope);
}
break;
default:
@@ -3374,45 +3430,16 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
// Validate NATURAL.
if (natural) {
- if (condition != null) {
- throw newValidationError(condition, RESOURCE.naturalDisallowsOnOrUsing());
+ if (join.getCondition() != null) {
+ throw newValidationError(getCondition(join), RESOURCE.naturalDisallowsOnOrUsing());
}
- // Join on fields that occur exactly once on each side. Ignore
- // fields that occur more than once on either side.
- final RelDataType leftRowType = getNamespaceOrThrow(left).getRowType();
- final RelDataType rightRowType = getNamespaceOrThrow(right).getRowType();
- final SqlNameMatcher nameMatcher = catalogReader.nameMatcher();
- List<String> naturalColumnNames =
- SqlValidatorUtil.deriveNaturalJoinColumnList(
- nameMatcher, leftRowType, rightRowType);
-
+ // Join on fields that occur on each side.
// Check compatibility of the chosen columns.
- for (String name : naturalColumnNames) {
- final RelDataType leftColType =
- requireNonNull(
- nameMatcher.field(leftRowType, name),
- () ->
- "unable to find left field "
- + name
- + " in "
- + leftRowType)
- .getType();
- final RelDataType rightColType =
- requireNonNull(
- nameMatcher.field(rightRowType, name),
- () ->
- "unable to find right field "
- + name
- + " in "
- + rightRowType)
- .getType();
- if (!SqlTypeUtil.isComparable(leftColType, rightColType)) {
- throw newValidationError(
- join,
- RESOURCE.naturalOrUsingColumnNotCompatible(
- name, leftColType.toString(), rightColType.toString()));
- }
+ for (String name : deriveNaturalJoinColumnList(join)) {
+ final SqlIdentifier id =
+ new SqlIdentifier(name, join.isNaturalNode().getParserPosition());
+ validateCommonJoinColumn(id, left, right, scope);
}
}
@@ -3430,13 +3457,13 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
case LEFT:
case RIGHT:
case FULL:
- if ((condition == null) && !natural) {
+ if ((join.getCondition() == null) && !natural) {
throw newValidationError(join, RESOURCE.joinRequiresCondition());
}
break;
case COMMA:
case CROSS:
- if (condition != null) {
+ if (join.getCondition() != null) {
throw newValidationError(
join.getConditionTypeNode(), RESOURCE.crossJoinDisallowsCondition());
}
@@ -3473,21 +3500,43 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
}
}
- private RelDataType validateUsingCol(SqlIdentifier id, SqlNode leftOrRight) {
- if (id.names.size() == 1) {
- String name = id.names.get(0);
- final SqlValidatorNamespace namespace = getNamespaceOrThrow(leftOrRight);
- final RelDataType rowType = namespace.getRowType();
- final SqlNameMatcher nameMatcher = catalogReader.nameMatcher();
- final RelDataTypeField field = nameMatcher.field(rowType, name);
- if (field != null) {
- if (nameMatcher.frequency(rowType.getFieldNames(), name) > 1) {
- throw newValidationError(id, RESOURCE.columnInUsingNotUnique(id.toString()));
- }
- return field.getType();
- }
+ /** Validates a column in a USING clause, or an inferred join key in a NATURAL join. */
+ private void validateCommonJoinColumn(
+ SqlIdentifier id, SqlNode left, SqlNode right, SqlValidatorScope scope) {
+ if (id.names.size() != 1) {
+ throw newValidationError(id, RESOURCE.columnNotFound(id.toString()));
+ }
+
+ final RelDataType leftColType = validateCommonInputJoinColumn(id, left, scope);
+ final RelDataType rightColType = validateCommonInputJoinColumn(id, right, scope);
+ if (!SqlTypeUtil.isComparable(leftColType, rightColType)) {
+ throw newValidationError(
+ id,
+ RESOURCE.naturalOrUsingColumnNotCompatible(
+ id.getSimple(), leftColType.toString(), rightColType.toString()));
}
- throw newValidationError(id, RESOURCE.columnNotFound(id.toString()));
+ }
+
+ /**
+ * Validates a column in a USING clause, or an inferred join key in a NATURAL join, in the left
+ * or right input to the join.
+ */
+ private RelDataType validateCommonInputJoinColumn(
+ SqlIdentifier id, SqlNode leftOrRight, SqlValidatorScope scope) {
+ Preconditions.checkArgument(id.names.size() == 1);
+ final String name = id.names.get(0);
+ final SqlValidatorNamespace namespace = getNamespaceOrThrow(leftOrRight);
+ final RelDataType rowType = namespace.getRowType();
+ final SqlNameMatcher nameMatcher = catalogReader.nameMatcher();
+ final RelDataTypeField field = nameMatcher.field(rowType, name);
+ if (field == null) {
+ throw newValidationError(id, RESOURCE.columnNotFound(name));
+ }
+ if (nameMatcher.frequency(rowType.getFieldNames(), name) > 1) {
+ throw newValidationError(id, RESOURCE.columnInUsingNotUnique(name));
+ }
+ checkRollUpInUsing(id, leftOrRight, scope);
+ return field.getType();
}
/**
@@ -3628,11 +3677,21 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
}
}
- private static @Nullable SqlNode stripDot(@Nullable SqlNode node) {
- if (node != null && node.getKind() == SqlKind.DOT) {
- return stripDot(((SqlCall) node).operand(0));
+ /**
+ * If the {@code node} is a DOT call, returns its first operand. Recurse, if the first operand
+ * is another DOT call.
+ *
+ * <p>In other words, it converts {@code a DOT b DOT c} to {@code a}.
+ *
+ * @param node The node to strip DOT
+ * @return the DOT's first operand
+ */
+ private static SqlNode stripDot(SqlNode node) {
+ SqlNode res = node;
+ while (res.getKind() == SqlKind.DOT) {
+ res = requireNonNull(((SqlCall) res).operand(0), "operand");
}
- return node;
+ return res;
}
private void checkRollUp(
@@ -3640,18 +3699,23 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
@Nullable SqlNode parent,
@Nullable SqlNode current,
SqlValidatorScope scope,
- @Nullable String optionalClause) {
+ @Nullable String contextClause) {
current = stripAs(current);
if (current instanceof SqlCall && !(current instanceof SqlSelect)) {
// Validate OVER separately
checkRollUpInWindow(getWindowInOver(current), scope);
current = stripOver(current);
- SqlNode stripDot = requireNonNull(stripDot(current), "stripDot(current)");
- List<? extends @Nullable SqlNode> children =
- ((SqlCall) stripAs(stripDot)).getOperandList();
- for (SqlNode child : children) {
- checkRollUp(parent, current, child, scope, optionalClause);
+ SqlNode stripDot = stripDot(current);
+ if (stripDot != current) {
+ // we stripped the field access. Recurse to this method, the DOT's operand
+ // can be another SqlCall, or an SqlIdentifier.
+ checkRollUp(grandParent, parent, stripDot, scope, contextClause);
+ } else {
+ List<? extends @Nullable SqlNode> children = ((SqlCall) stripDot).getOperandList();
+ for (SqlNode child : children) {
+ checkRollUp(parent, current, child, scope, contextClause);
+ }
}
} else if (current instanceof SqlIdentifier) {
SqlIdentifier id = (SqlIdentifier) current;
@@ -3660,7 +3724,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
|| !isRolledUpColumnAllowedInAgg(
id, scope, (SqlCall) parent, grandParent)) {
String context =
- optionalClause != null ? optionalClause : parent.getKind().toString();
+ contextClause != null ? contextClause : parent.getKind().toString();
throw newValidationError(
id, RESOURCE.rolledUpNotAllowed(deriveAliasNonNull(id, 0), context));
}
@@ -4113,10 +4177,16 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
private void validateGroupByExpr(SqlNode groupByItem, SqlValidatorScope groupByScope) {
switch (groupByItem.getKind()) {
+ case GROUP_BY_DISTINCT:
+ SqlCall call = (SqlCall) groupByItem;
+ for (SqlNode operand : call.getOperandList()) {
+ validateGroupByExpr(operand, groupByScope);
+ }
+ break;
case GROUPING_SETS:
case ROLLUP:
case CUBE:
- final SqlCall call = (SqlCall) groupByItem;
+ call = (SqlCall) groupByItem;
for (SqlNode operand : call.getOperandList()) {
validateExpr(operand, groupByScope);
}
@@ -4190,6 +4260,7 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
// expressions, because they do not have a type.
for (SqlNode node : groupList) {
switch (node.getKind()) {
+ case GROUP_BY_DISTINCT:
case GROUPING_SETS:
case ROLLUP:
case CUBE:
@@ -4226,6 +4297,11 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
@Nullable AggregatingSelectScope aggregatingScope,
SqlNode groupItem) {
switch (groupItem.getKind()) {
+ case GROUP_BY_DISTINCT:
+ for (SqlNode sqlNode : ((SqlCall) groupItem).getOperandList()) {
+ validateGroupItem(groupScope, aggregatingScope, sqlNode);
+ }
+ break;
case GROUPING_SETS:
case ROLLUP:
case CUBE:
@@ -4267,11 +4343,23 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
condition.validate(this, scope);
final RelDataType type = deriveType(scope, condition);
- if (!SqlTypeUtil.inBooleanFamily(type)) {
+ if (!isReturnBooleanType(type)) {
throw newValidationError(condition, RESOURCE.condMustBeBoolean(clause));
}
}
+ private static boolean isReturnBooleanType(RelDataType relDataType) {
+ if (relDataType instanceof RelRecordType) {
+ RelRecordType recordType = (RelRecordType) relDataType;
+ Preconditions.checkState(
+ recordType.getFieldList().size() == 1,
+ "sub-query as condition must return only one column");
+ RelDataTypeField recordField = recordType.getFieldList().get(0);
+ return SqlTypeUtil.inBooleanFamily(recordField.getType());
+ }
+ return SqlTypeUtil.inBooleanFamily(relDataType);
+ }
+
protected void validateHavingClause(SqlSelect select) {
// HAVING is validated in the scope after groups have been created.
// For example, in "SELECT empno FROM emp WHERE empno = 10 GROUP BY
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/RelDecorrelator.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/RelDecorrelator.java
index 8e88e0addd2..d750a561236 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/RelDecorrelator.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql2rel/RelDecorrelator.java
@@ -503,13 +503,10 @@ public class RelDecorrelator implements ReflectiveVisitor {
RelCollation oldCollation = rel.getCollation();
RelCollation newCollation = RexUtil.apply(mapping, oldCollation);
- final int offset = rel.offset == null ? -1 : RexLiteral.intValue(rel.offset);
- final int fetch = rel.fetch == null ? -1 : RexLiteral.intValue(rel.fetch);
-
final RelNode newSort =
relBuilder
.push(newInput)
- .sortLimit(offset, fetch, relBuilder.fields(newCollation))
+ .sortLimit(rel.offset, rel.fetch, relBuilder.fields(newCollation))
.build();
// Sort does not change input ordering
@@ -561,20 +558,21 @@ public class RelDecorrelator implements ReflectiveVisitor {
int newPos = 0;
- // oldInput has the original group by keys in the front.
+ final List<Integer> groupKeyIndices = rel.getGroupSet().asList();
final NavigableMap<Integer, RexLiteral> omittedConstants = new TreeMap<>();
for (int i = 0; i < oldGroupKeyCount; i++) {
- final RexLiteral constant = projectedLiteral(newInput, i);
+ final int idx = groupKeyIndices.get(i);
+ final RexLiteral constant = projectedLiteral(newInput, idx);
if (constant != null) {
// Exclude constants. Aggregate({true}) occurs because Aggregate({})
// would generate 1 row even when applied to an empty table.
- omittedConstants.put(i, constant);
+ omittedConstants.put(idx, constant);
continue;
}
// add mapping of group keys.
- outputMap.put(i, newPos);
- int newInputPos = frame.oldToNewOutputs.get(i);
+ outputMap.put(idx, newPos);
+ int newInputPos = requireNonNull(frame.oldToNewOutputs.get(idx));
projects.add(RexInputRef.of2(newInputPos, newInputOutput));
mapNewInputToProjOutputs.put(newInputPos, newPos);
newPos++;
@@ -616,8 +614,6 @@ public class RelDecorrelator implements ReflectiveVisitor {
.projectNamed(Pair.left(projects), Pair.right(projects), true)
.build();
- newProject = RelOptUtil.copyRelHints(newInput, newProject);
-
// update mappings:
// oldInput ----> newInput
//
@@ -717,7 +713,7 @@ public class RelDecorrelator implements ReflectiveVisitor {
relBuilder.project(postProjects);
}
- RelNode newRel = RelOptUtil.copyRelHints(rel, relBuilder.build());
+ RelNode newRel = relBuilder.build();
// Aggregate does not change input ordering so corVars will be
// located at the same position as the input newProject.
@@ -831,8 +827,6 @@ public class RelDecorrelator implements ReflectiveVisitor {
.projectNamed(Pair.left(projects), Pair.right(projects), true)
.build();
- newProject = RelOptUtil.copyRelHints(rel, newProject);
-
return register(rel, newProject, mapOldToNewOutputs, corDefOutputs);
}
@@ -870,7 +864,7 @@ public class RelDecorrelator implements ReflectiveVisitor {
newLocalOutputs = mapNewInputToOutputs.get(newInput);
}
- final int newCorVarOffset = frame.oldToNewOutputs.get(oldCorVarOffset);
+ final int newCorVarOffset = requireNonNull(frame.oldToNewOutputs.get(oldCorVarOffset));
// Add all unique positions referenced.
if (!newLocalOutputs.contains(newCorVarOffset)) {
@@ -946,7 +940,7 @@ public class RelDecorrelator implements ReflectiveVisitor {
mapNewInputToOutputs.get(newInput),
() -> "mapNewInputToOutputs.get(" + newInput + ")");
- final int newLocalOutput = frame.oldToNewOutputs.get(corRef.field);
+ final int newLocalOutput = requireNonNull(frame.oldToNewOutputs.get(corRef.field));
// newOutput is the index of the corVar in the referenced
// position list plus the offset of referenced position list of
@@ -1276,7 +1270,7 @@ public class RelDecorrelator implements ReflectiveVisitor {
if (!corDef.corr.equals(rel.getCorrelationId())) {
continue;
}
- final int newLeftPos = leftFrame.oldToNewOutputs.get(corDef.field);
+ final int newLeftPos = requireNonNull(leftFrame.oldToNewOutputs.get(corDef.field));
final int newRightPos = rightOutput.getValue();
conditions.add(
relBuilder.equals(
@@ -1315,7 +1309,8 @@ public class RelDecorrelator implements ReflectiveVisitor {
// Right input positions are shifted by newLeftFieldCount.
for (int i = 0; i < oldRightFieldCount; i++) {
mapOldToNewOutputs.put(
- i + oldLeftFieldCount, rightFrame.oldToNewOutputs.get(i) + newLeftFieldCount);
+ i + oldLeftFieldCount,
+ requireNonNull(rightFrame.oldToNewOutputs.get(i)) + newLeftFieldCount);
}
final RexNode condition =
@@ -1368,7 +1363,6 @@ public class RelDecorrelator implements ReflectiveVisitor {
decorrelateExpr(
castNonNull(currentRel), map, cm, rel.getCondition()),
ImmutableSet.of())
- .hints(rel.getHints())
.build();
// Create the mapping between the output of the old correlation rel
@@ -1388,7 +1382,8 @@ public class RelDecorrelator implements ReflectiveVisitor {
// Right input positions are shifted by newLeftFieldCount.
for (int i = 0; i < oldRightFieldCount; i++) {
mapOldToNewOutputs.put(
- i + oldLeftFieldCount, rightFrame.oldToNewOutputs.get(i) + newLeftFieldCount);
+ i + oldLeftFieldCount,
+ requireNonNull(rightFrame.oldToNewOutputs.get(i)) + newLeftFieldCount);
}
final NavigableMap<CorDef, Integer> corDefOutputs = new TreeMap<>(leftFrame.corDefOutputs);
@@ -1437,7 +1432,7 @@ public class RelDecorrelator implements ReflectiveVisitor {
int newLocalOrdinal = oldLocalOrdinal;
if (!frame.oldToNewOutputs.isEmpty()) {
- newLocalOrdinal = frame.oldToNewOutputs.get(oldLocalOrdinal);
+ newLocalOrdinal = requireNonNull(frame.oldToNewOutputs.get(oldLocalOrdinal));
}
newOrdinal += newLocalOrdinal;
@@ -1646,6 +1641,7 @@ public class RelDecorrelator implements ReflectiveVisitor {
RelNode newRel,
Map<Integer, Integer> oldToNewOutputs,
NavigableMap<CorDef, Integer> corDefOutputs) {
+ newRel = RelOptUtil.copyRelHints(rel, newRel);
final Frame frame = new Frame(rel, newRel, corDefOutputs, oldToNewOutputs);
map.put(rel, frame);
return frame;
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 2b0844071c3..70c37411e82 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
@@ -25,7 +25,9 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import org.apache.calcite.avatica.util.Spaces;
+import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.linq4j.Ord;
+import org.apache.calcite.linq4j.tree.TableExpressionFactory;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelOptSamplingParameters;
@@ -38,11 +40,14 @@ import org.apache.calcite.prepare.RelOptTableImpl;
import org.apache.calcite.rel.RelCollation;
import org.apache.calcite.rel.RelCollationTraitDef;
import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelDistributions;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelRoot;
import org.apache.calcite.rel.RelShuttleImpl;
import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.Collect;
import org.apache.calcite.rel.core.CorrelationId;
@@ -97,6 +102,7 @@ import org.apache.calcite.rex.RexWindowBounds;
import org.apache.calcite.schema.ColumnStrategy;
import org.apache.calcite.schema.ModifiableTable;
import org.apache.calcite.schema.ModifiableView;
+import org.apache.calcite.schema.Schemas;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.TranslatableTable;
import org.apache.calcite.schema.Wrapper;
@@ -219,10 +225,10 @@ import static org.apache.calcite.sql.SqlUtil.stripAs;
* <p>FLINK modifications are at lines
*
* <ol>
- * <li>Added in FLINK-29081, FLINK-28682: Lines 633 ~ 643
- * <li>Added in FLINK-28682: Lines 2130 ~ 2147
- * <li>Added in FLINK-28682: Lines 2184 ~ 2212
- * <li>Added in FLINK-20873: Lines 5245 ~ 5254
+ * <li>Added in FLINK-29081, FLINK-28682: Lines 644 ~ 654
+ * <li>Added in FLINK-28682: Lines 2277 ~ 2294
+ * <li>Added in FLINK-28682: Lines 2331 ~ 2359
+ * <li>Added in FLINK-20873: Lines 5431 ~ 5440
* </ol>
*/
@SuppressWarnings("UnstableApiUsage")
@@ -627,6 +633,11 @@ public class SqlToRelConverter {
hints = SqlUtil.getRelHint(hintStrategies, select.getHints());
}
}
+
+ if (config.isAddJsonTypeOperatorEnabled()) {
+ result = result.accept(new NestedJsonFunctionRelRewriter());
+ }
+
// propagate the hints.
result = RelOptUtil.propagateRelHints(result, false);
@@ -1403,12 +1414,143 @@ public class SqlToRelConverter {
//
bb.cursors.add(converted.r);
return;
-
+ case SET_SEMANTICS_TABLE:
+ if (!config.isExpand()) {
+ return;
+ }
+ substituteSubQueryOfSetSemanticsInputTable(bb, subQuery);
+ return;
default:
throw new AssertionError("unexpected kind of sub-query: " + subQuery.node);
}
}
+ private void substituteSubQueryOfSetSemanticsInputTable(Blackboard bb, SubQuery subQuery) {
+ SqlBasicCall call;
+ SqlNode query;
+ call = (SqlBasicCall) subQuery.node;
+ query = call.operand(0);
+ final SqlValidatorScope innerTableScope =
+ (query instanceof SqlSelect) ? validator().getSelectScope((SqlSelect) query) : null;
+ final Blackboard setSemanticsTableBb = createBlackboard(innerTableScope, null, false);
+ final RelNode inputOfSetSemanticsTable =
+ convertQueryRecursive(query, false, null).project();
+ requireNonNull(inputOfSetSemanticsTable, () -> "input RelNode is null for query " + query);
+ SqlNodeList partitionList = call.operand(1);
+ final ImmutableBitSet partitionKeys =
+ buildPartitionKeys(setSemanticsTableBb, partitionList);
+ // For set semantics table, distribution is singleton if does not specify
+ // partition keys
+ RelDistribution distribution =
+ partitionKeys.isEmpty()
+ ? RelDistributions.SINGLETON
+ : RelDistributions.hash(partitionKeys.asList());
+ // ORDER BY
+ final SqlNodeList orderList = call.operand(2);
+ final RelCollation orders = buildCollation(setSemanticsTableBb, orderList);
+ relBuilder.push(inputOfSetSemanticsTable);
+ if (orderList.isEmpty()) {
+ relBuilder.exchange(distribution);
+ } else {
+ relBuilder.sortExchange(distribution, orders);
+ }
+ RelNode tableRel = relBuilder.build();
+ subQuery.expr = bb.register(tableRel, JoinRelType.LEFT);
+ // This is used when converting window table functions:
+ //
+ // select * from table(tumble(table emps, descriptor(deptno),
+ // interval '3' DAY))
+ //
+ bb.cursors.add(tableRel);
+ }
+
+ private ImmutableBitSet buildPartitionKeys(Blackboard bb, SqlNodeList partitionList) {
+ final ImmutableBitSet.Builder partitionKeys = ImmutableBitSet.builder();
+ for (SqlNode partition : partitionList) {
+ validator().deriveType(bb.scope(), partition);
+ RexNode e = bb.convertExpression(partition);
+ partitionKeys.set(parseFieldIdx(e));
+ }
+ return partitionKeys.build();
+ }
+
+ /**
+ * Note: The ORDER BY clause for input table parameter differs from the ORDER BY clause in some
+ * other contexts in that only columns may be sorted (not arbitrary expressions).
+ *
+ * @param bb Scope within which to resolve identifiers
+ * @param orderList Order by clause, may be null
+ * @return ordering of input table
+ */
+ private RelCollation buildCollation(Blackboard bb, SqlNodeList orderList) {
+ final List<RelFieldCollation> orderKeys = new ArrayList<>();
+ for (SqlNode orderItem : orderList) {
+ orderKeys.add(
+ convertOrderItem(
+ bb,
+ orderItem,
+ RelFieldCollation.Direction.ASCENDING,
+ RelFieldCollation.NullDirection.UNSPECIFIED));
+ }
+ return cluster.traitSet().canonize(RelCollations.of(orderKeys));
+ }
+
+ private RelFieldCollation convertOrderItem(
+ Blackboard bb,
+ SqlNode orderItem,
+ RelFieldCollation.Direction direction,
+ RelFieldCollation.NullDirection nullDirection) {
+ switch (orderItem.getKind()) {
+ case DESCENDING:
+ return convertOrderItem(
+ bb,
+ ((SqlCall) orderItem).operand(0),
+ RelFieldCollation.Direction.DESCENDING,
+ nullDirection);
+ case NULLS_FIRST:
+ return convertOrderItem(
+ bb,
+ ((SqlCall) orderItem).operand(0),
+ direction,
+ RelFieldCollation.NullDirection.FIRST);
+ case NULLS_LAST:
+ return convertOrderItem(
+ bb,
+ ((SqlCall) orderItem).operand(0),
+ direction,
+ RelFieldCollation.NullDirection.LAST);
+ default:
+ break;
+ }
+
+ switch (nullDirection) {
+ case UNSPECIFIED:
+ nullDirection =
+ validator().config().defaultNullCollation().last(desc(direction))
+ ? RelFieldCollation.NullDirection.LAST
+ : RelFieldCollation.NullDirection.FIRST;
+ break;
+ default:
+ break;
+ }
+
+ RexNode e = bb.convertExpression(orderItem);
+ return new RelFieldCollation(parseFieldIdx(e), direction, nullDirection);
+ }
+
+ private static int parseFieldIdx(RexNode e) {
+ switch (e.getKind()) {
+ case FIELD_ACCESS:
+ final RexFieldAccess f = (RexFieldAccess) e;
+ return f.getField().getIndex();
+ case INPUT_REF:
+ final RexInputRef ref = (RexInputRef) e;
+ return ref.getIndex();
+ default:
+ throw new AssertionError();
+ }
+ }
+
private RexNode translateIn(RelOptUtil.Logic logic, @Nullable RelNode root, final RexNode rex) {
switch (logic) {
case TRUE:
@@ -1430,15 +1572,19 @@ public class SqlToRelConverter {
//
// The first two lists contain the same number of fields.
final int k = (fieldCount - 1) / 2;
+ ImmutableList.Builder<RexNode> rexNodeBuilder = ImmutableList.builder();
+ rexNodeBuilder.add(rexNode);
for (int i = 0; i < k; i++) {
- rexNode =
+ rexNodeBuilder.add(
rexBuilder.makeCall(
- SqlStdOperatorTable.AND,
- rexNode,
- rexBuilder.makeCall(
- SqlStdOperatorTable.IS_NOT_NULL,
- rexBuilder.makeFieldAccess(rex, i)));
+ SqlStdOperatorTable.IS_NOT_NULL,
+ rexBuilder.makeFieldAccess(rex, i)));
}
+ rexNode =
+ rexBuilder.makeCall(
+ rexNode.getType(),
+ SqlStdOperatorTable.AND,
+ RexUtil.flatten(rexNodeBuilder.build(), SqlStdOperatorTable.AND));
return rexNode;
case TRUE_FALSE_UNKNOWN:
@@ -1735,8 +1881,8 @@ public class SqlToRelConverter {
final RelDataType rowType;
if (targetRowType != null) {
rowType =
- typeFactory.createTypeWithNullability(
- targetRowType, SqlTypeUtil.containsNullable(listType));
+ SqlTypeUtil.keepSourceTypeAndTargetNullability(
+ targetRowType, listType, typeFactory);
} else {
rowType = SqlTypeUtil.promoteToRowType(typeFactory, listType, null);
}
@@ -1880,6 +2026,7 @@ public class SqlToRelConverter {
case ARRAY_QUERY_CONSTRUCTOR:
case MAP_QUERY_CONSTRUCTOR:
case CURSOR:
+ case SET_SEMANTICS_TABLE:
case SCALAR_QUERY:
if (!registerOnlyScalarSubQueries || (kind == SqlKind.SCALAR_QUERY)) {
bb.registerSubQuery(node, RelOptUtil.Logic.TRUE_FALSE);
@@ -2370,13 +2517,10 @@ public class SqlToRelConverter {
// PARTITION BY
final SqlNodeList partitionList = matchRecognize.getPartitionList();
- final ImmutableBitSet.Builder partitionKeys = ImmutableBitSet.builder();
- for (SqlNode partition : partitionList) {
- RexNode e = matchBb.convertExpression(partition);
- partitionKeys.set(((RexInputRef) e).getIndex());
- }
+ final ImmutableBitSet partitionKeys = buildPartitionKeys(matchBb, partitionList);
// ORDER BY
+ // TODO combine with buildCollation method after support NULLS_FIRST/NULLS_LAST
final SqlNodeList orderList = matchRecognize.getOrderList();
final List<RelFieldCollation> orderKeys = new ArrayList<>();
for (SqlNode order : orderList) {
@@ -2520,7 +2664,7 @@ public class SqlToRelConverter {
after,
subsetMap,
allRows,
- partitionKeys.build(),
+ partitionKeys,
orders,
intervalNode);
bb.setRoot(rel, false);
@@ -2731,8 +2875,19 @@ public class SqlToRelConverter {
final SqlUserDefinedTableMacro udf = (SqlUserDefinedTableMacro) operator;
final TranslatableTable table = udf.getTable(callBinding);
final RelDataType rowType = table.getRowType(typeFactory);
+ CalciteSchema schema =
+ Schemas.subSchema(
+ catalogReader.getRootSchema(), udf.getNameAsId().skipLast(1).names);
+ TableExpressionFactory expressionFunction =
+ clazz ->
+ Schemas.getTableExpression(
+ Objects.requireNonNull(schema, "schema").plus(),
+ Util.last(udf.getNameAsId().names),
+ table,
+ clazz);
RelOptTable relOptTable =
- RelOptTableImpl.create(null, rowType, table, udf.getNameAsId().names);
+ RelOptTableImpl.create(
+ null, rowType, udf.getNameAsId().names, table, expressionFunction);
RelNode converted = toRel(relOptTable, ImmutableList.of());
bb.setRoot(converted, true);
return;
@@ -2759,15 +2914,6 @@ public class SqlToRelConverter {
validator().getValidatedNodeType(call),
columnMappings);
- final SqlValidatorScope selectScope = ((DelegatingScope) bb.scope()).getParent();
- final Blackboard seekBb = createBlackboard(selectScope, null, false);
-
- final CorrelationUse p = getCorrelationUse(seekBb, callRel);
- if (p != null) {
- assert p.r instanceof LogicalTableFunctionScan;
- callRel = (LogicalTableFunctionScan) p.r;
- }
-
bb.setRoot(callRel, true);
afterTableFunction(bb, call, callRel);
}
@@ -3099,7 +3245,9 @@ public class SqlToRelConverter {
rightRel,
condition,
convertJoinType(join.getJoinType()));
- bb.setRoot(joinRel, false);
+ relBuilder.push(joinRel);
+ final RelNode newProjectRel = relBuilder.project(relBuilder.fields()).build();
+ bb.setRoot(newProjectRel, false);
}
private RexNode convertNaturalCondition(
@@ -3169,6 +3317,15 @@ public class SqlToRelConverter {
bb.root == null || bb.registered.size() == 0 ? rightRel : bb.reRegister(rightRel);
bb.setRoot(ImmutableList.of(leftRel, newRightRel));
RexNode conditionExp = bb.convertExpression(condition);
+ if (conditionExp instanceof RexInputRef && newRightRel != rightRel) {
+ int leftFieldCount = leftRel.getRowType().getFieldCount();
+ List<RelDataTypeField> rightFieldList = newRightRel.getRowType().getFieldList();
+ int rightFieldCount = newRightRel.getRowType().getFieldCount();
+ conditionExp =
+ rexBuilder.makeInputRef(
+ rightFieldList.get(rightFieldCount - 1).getType(),
+ leftFieldCount + rightFieldCount - 1);
+ }
return Pair.of(conditionExp, newRightRel);
}
@@ -4437,7 +4594,18 @@ public class SqlToRelConverter {
fieldNames, catalogReader.nameMatcher().isCaseSensitive());
relBuilder.push(bb.root()).projectNamed(exprs, fieldNames, true);
- bb.setRoot(relBuilder.build(), false);
+
+ RelNode project = relBuilder.build();
+
+ final RelNode r;
+ final CorrelationUse p = getCorrelationUse(bb, project);
+ if (p != null) {
+ r = p.r;
+ } else {
+ r = project;
+ }
+
+ bb.setRoot(r, false);
assert bb.columnMonotonicities.isEmpty();
bb.columnMonotonicities.addAll(columnMonotonicityList);
@@ -4747,7 +4915,17 @@ public class SqlToRelConverter {
List<RegisterArgs> registerCopy = registered;
registered = new ArrayList<>();
for (RegisterArgs reg : registerCopy) {
- register(reg.rel, reg.joinType, reg.leftKeys);
+ RelNode relNode = reg.rel;
+ relBuilder.push(relNode);
+ final RelMetadataQuery mq = relBuilder.getCluster().getMetadataQuery();
+ final Boolean unique = mq.areColumnsUnique(relBuilder.peek(), ImmutableBitSet.of());
+ if (unique == null || !unique) {
+ relBuilder.aggregate(
+ relBuilder.groupKey(),
+ relBuilder.aggregateCall(
+ SqlStdOperatorTable.SINGLE_VALUE, relBuilder.field(0)));
+ }
+ register(relBuilder.build(), reg.joinType, reg.leftKeys);
}
return requireNonNull(this.root, "root");
}
@@ -4862,7 +5040,15 @@ public class SqlToRelConverter {
int i = 0;
int offset = 0;
for (SqlValidatorNamespace c : ancestorScope1.getChildren()) {
- builder.addAll(c.getRowType().getFieldList());
+ if (ancestorScope1.isChildNullable(i)) {
+ for (final RelDataTypeField f : c.getRowType().getFieldList()) {
+ builder.add(
+ f.getName(),
+ typeFactory.createTypeWithNullability(f.getType(), true));
+ }
+ } else {
+ builder.addAll(c.getRowType().getFieldList());
+ }
if (i == resolve.path.steps().get(0).i) {
for (RelDataTypeField field : c.getRowType().getFieldList()) {
fields.put(field.getName(), field.getIndex() + offset);
@@ -6480,5 +6666,140 @@ public class SqlToRelConverter {
/** Sets {@link #getHintStrategyTable()}. */
Config withHintStrategyTable(HintStrategyTable hintStrategyTable);
+
+ /**
+ * Whether add {@link SqlStdOperatorTable#JSON_TYPE_OPERATOR} for between json functions.
+ */
+ @Value.Default
+ default boolean isAddJsonTypeOperatorEnabled() {
+ return true;
+ }
+
+ /** Sets {@link #isAddJsonTypeOperatorEnabled()}. */
+ Config withAddJsonTypeOperatorEnabled(boolean addJsonTypeOperatorEnabled);
+ }
+
+ /**
+ * Used to find nested json functions, and add {@link SqlStdOperatorTable#JSON_TYPE_OPERATOR} to
+ * nested json output.
+ */
+ private class NestedJsonFunctionRelRewriter extends RelShuttleImpl {
+
+ @Override
+ public RelNode visit(LogicalProject project) {
+ final Set<Integer> jsonInputFields = findJsonInputs(project.getInput());
+ final Set<Integer> requiredJsonFieldsFromParent =
+ stack.size() > 0
+ ? requiredJsonOutputFromParent(stack.getLast())
+ : Collections.emptySet();
+
+ final List<RexNode> originalProjections = project.getProjects();
+ final ImmutableList.Builder<RexNode> newProjections = ImmutableList.builder();
+ JsonFunctionRexRewriter rexRewriter = new JsonFunctionRexRewriter(jsonInputFields);
+ for (int i = 0; i < originalProjections.size(); ++i) {
+ if (requiredJsonFieldsFromParent.contains(i)) {
+ newProjections.add(rexRewriter.forceChildJsonType(originalProjections.get(i)));
+ } else {
+ newProjections.add(originalProjections.get(i).accept(rexRewriter));
+ }
+ }
+
+ RelNode newInput = project.getInput().accept(this);
+ return LogicalProject.create(
+ newInput,
+ project.getHints(),
+ newProjections.build(),
+ project.getRowType().getFieldNames());
+ }
+
+ private Set<Integer> requiredJsonOutputFromParent(RelNode relNode) {
+ if (!(relNode instanceof Aggregate)) {
+ return Collections.emptySet();
+ }
+ final Aggregate aggregate = (Aggregate) relNode;
+ final List<AggregateCall> aggregateCalls = aggregate.getAggCallList();
+ final ImmutableSet.Builder<Integer> result = ImmutableSet.builder();
+ for (final AggregateCall call : aggregateCalls) {
+ if (call.getAggregation() == SqlStdOperatorTable.JSON_OBJECTAGG) {
+ result.add(call.getArgList().get(1));
+ } else if (call.getAggregation() == SqlStdOperatorTable.JSON_ARRAYAGG) {
+ result.add(call.getArgList().get(0));
+ }
+ }
+ return result.build();
+ }
+
+ private Set<Integer> findJsonInputs(RelNode relNode) {
+ if (!(relNode instanceof Aggregate)) {
+ return Collections.emptySet();
+ }
+ final Aggregate aggregate = (Aggregate) relNode;
+ final List<AggregateCall> aggregateCalls = aggregate.getAggCallList();
+ final ImmutableSet.Builder<Integer> result = ImmutableSet.builder();
+ for (int i = 0; i < aggregateCalls.size(); ++i) {
+ final AggregateCall call = aggregateCalls.get(i);
+ if (call.getAggregation() == SqlStdOperatorTable.JSON_OBJECTAGG
+ || call.getAggregation() == SqlStdOperatorTable.JSON_ARRAYAGG) {
+ result.add(aggregate.getGroupCount() + i);
+ }
+ }
+ return result.build();
+ }
+ }
+
+ /** Used to rewrite json functions which is nested. */
+ private class JsonFunctionRexRewriter extends RexShuttle {
+
+ private final Set<Integer> jsonInputFields;
+
+ JsonFunctionRexRewriter(Set<Integer> jsonInputFields) {
+ this.jsonInputFields = jsonInputFields;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ if (call.getOperator() == SqlStdOperatorTable.JSON_OBJECT) {
+ final ImmutableList.Builder<RexNode> builder = ImmutableList.builder();
+ for (int i = 0; i < call.operands.size(); ++i) {
+ if ((i & 1) == 0 && i != 0) {
+ builder.add(forceChildJsonType(call.operands.get(i)));
+ } else {
+ builder.add(call.operands.get(i));
+ }
+ }
+ return rexBuilder.makeCall(SqlStdOperatorTable.JSON_OBJECT, builder.build());
+ }
+ if (call.getOperator() == SqlStdOperatorTable.JSON_ARRAY) {
+ final ImmutableList.Builder<RexNode> builder = ImmutableList.builder();
+ builder.add(call.operands.get(0));
+ for (int i = 1; i < call.operands.size(); ++i) {
+ builder.add(forceChildJsonType(call.operands.get(i)));
+ }
+ return rexBuilder.makeCall(SqlStdOperatorTable.JSON_ARRAY, builder.build());
+ }
+ return super.visitCall(call);
+ }
+
+ private RexNode forceChildJsonType(RexNode rexNode) {
+ final RexNode childResult = rexNode.accept(this);
+ if (isJsonResult(rexNode)) {
+ return rexBuilder.makeCall(SqlStdOperatorTable.JSON_TYPE_OPERATOR, childResult);
+ }
+ return childResult;
+ }
+
+ private boolean isJsonResult(RexNode rexNode) {
+ if (rexNode instanceof RexCall) {
+ final RexCall call = (RexCall) rexNode;
+ final SqlOperator operator = call.getOperator();
+ return operator == SqlStdOperatorTable.JSON_OBJECT
+ || operator == SqlStdOperatorTable.JSON_ARRAY
+ || operator == SqlStdOperatorTable.JSON_VALUE;
+ } else if (rexNode instanceof RexInputRef) {
+ final RexInputRef inputRef = (RexInputRef) rexNode;
+ return jsonInputFields.contains(inputRef.getIndex());
+ }
+ return false;
+ }
}
}
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 00e5075e3c5..3401942c57f 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
@@ -77,6 +77,7 @@ import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexCallBinding;
import org.apache.calcite.rex.RexCorrelVariable;
+import org.apache.calcite.rex.RexDynamicParam;
import org.apache.calcite.rex.RexExecutor;
import org.apache.calcite.rex.RexFieldCollation;
import org.apache.calcite.rex.RexInputRef;
@@ -138,6 +139,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
@@ -154,7 +156,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 2928 ~ 2931
+ * <li>Should be removed after fix of FLINK-29804: Lines 2945 ~ 2948
* </ol>
*/
@Value.Enclosing
@@ -162,7 +164,7 @@ public class RelBuilder {
protected final RelOptCluster cluster;
protected final @Nullable RelOptSchema relOptSchema;
private final Deque<Frame> stack = new ArrayDeque<>();
- private final RexSimplify simplifier;
+ private RexSimplify simplifier;
private final Config config;
private final RelOptTable.ViewExpander viewExpander;
private RelFactories.Struct struct;
@@ -428,6 +430,25 @@ public class RelBuilder {
}
}
+ /** Performs an action with a temporary simplifier. */
+ public <E> E withSimplifier(
+ BiFunction<RelBuilder, RexSimplify, RexSimplify> simplifierTransform,
+ Function<RelBuilder, E> fn) {
+ final RexSimplify previousSimplifier = this.simplifier;
+ try {
+ this.simplifier = simplifierTransform.apply(this, previousSimplifier);
+ return fn.apply(this);
+ } finally {
+ this.simplifier = previousSimplifier;
+ }
+ }
+
+ /** Performs an action using predicates of the {@link #peek() current node} to simplify. */
+ public <E> E withPredicates(RelMetadataQuery mq, Function<RelBuilder, E> fn) {
+ final RelOptPredicateList predicates = mq.getPulledUpPredicates(peek());
+ return withSimplifier((r, s) -> s.withPredicates(predicates), fn);
+ }
+
// Methods that return scalar expressions
/** Creates a literal (constant expression). */
@@ -2107,12 +2128,8 @@ public class RelBuilder {
// Simplify expressions.
if (config.simplify()) {
- final RexShuttle shuttle = RexUtil.searchShuttle(getRexBuilder(), null, 2);
for (int i = 0; i < nodeList.size(); i++) {
- final RexNode node0 = nodeList.get(i);
- final RexNode node1 = simplifier.simplifyPreservingType(node0);
- final RexNode node2 = node1.accept(shuttle);
- nodeList.set(i, node2);
+ nodeList.set(i, simplifier.simplifyPreservingType(nodeList.get(i)));
}
}
@@ -3389,11 +3406,40 @@ public class RelBuilder {
* @param nodes Sort expressions
*/
public RelBuilder sortLimit(int offset, int fetch, Iterable<? extends RexNode> nodes) {
+ final @Nullable RexNode offsetNode = offset <= 0 ? null : literal(offset);
+ final @Nullable RexNode fetchNode = fetch < 0 ? null : literal(fetch);
+ return sortLimit(offsetNode, fetchNode, nodes);
+ }
+
+ /**
+ * Creates a {@link Sort} by a list of expressions, with limitNode and offsetNode.
+ *
+ * @param offsetNode RexLiteral means number of rows to skip is deterministic, RexDynamicParam
+ * means number of rows to skip is dynamic.
+ * @param fetchNode RexLiteral means maximum number of rows to fetch is deterministic,
+ * RexDynamicParam mean maximum number is dynamic.
+ * @param nodes Sort expressions
+ */
+ public RelBuilder sortLimit(
+ @Nullable RexNode offsetNode,
+ @Nullable RexNode fetchNode,
+ Iterable<? extends RexNode> nodes) {
+ if (offsetNode != null) {
+ if (!(offsetNode instanceof RexLiteral || offsetNode instanceof RexDynamicParam)) {
+ throw new IllegalArgumentException(
+ "OFFSET node must be RexLiteral or RexDynamicParam");
+ }
+ }
+ if (fetchNode != null) {
+ if (!(fetchNode instanceof RexLiteral || fetchNode instanceof RexDynamicParam)) {
+ throw new IllegalArgumentException(
+ "FETCH node must be RexLiteral or RexDynamicParam");
+ }
+ }
+
final Registrar registrar = new Registrar(fields(), ImmutableList.of());
final List<RelFieldCollation> fieldCollations = registrar.registerFieldCollations(nodes);
-
- final RexNode offsetNode = offset <= 0 ? null : literal(offset);
- final RexNode fetchNode = fetch < 0 ? null : literal(fetch);
+ final int fetch = fetchNode instanceof RexLiteral ? RexLiteral.intValue(fetchNode) : -1;
if (offsetNode == null && fetch == 0 && config.simplifyLimit()) {
return empty();
}
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 bdeec088414..e4838a5cf29 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
@@ -6,15 +6,15 @@ 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.google.guava:guava:29.0-jre
+- com.google.guava:guava:31.1-jre
- com.google.guava:failureaccess:1.0.1
- com.esri.geometry:esri-geometry-api:2.2.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
+- org.apache.calcite:calcite-core:1.31.0
+- org.apache.calcite:calcite-linq4j:1.31.0
+- org.apache.calcite.avatica:avatica-core:1.22.0
- commons-codec:commons-codec:1.15
- commons-io:commons-io:2.11.0
This project bundles the following dependencies under the MIT License. (http://www.opensource.org/licenses/mit-license.php)
-- org.checkerframework:checker-qual:3.10.0
+- org.checkerframework:checker-qual:3.12.0
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
index b6a364800cf..1f7e77d707c 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala
@@ -65,7 +65,8 @@ class FlinkPlannerImpl(
val operatorTable: SqlOperatorTable = config.getOperatorTable
val parser: CalciteParser = new CalciteParser(config.getParserConfig)
val convertletTable: SqlRexConvertletTable = config.getConvertletTable
- val sqlToRelConverterConfig: SqlToRelConverter.Config = config.getSqlToRelConverterConfig
+ val sqlToRelConverterConfig: SqlToRelConverter.Config =
+ config.getSqlToRelConverterConfig.withAddJsonTypeOperatorEnabled(false)
var validator: FlinkCalciteSqlValidator = _
@@ -75,7 +76,7 @@ class FlinkPlannerImpl(
catalogReaderSupplier.apply(true), // ignore cases for lenient completion
typeFactory,
SqlValidator.Config.DEFAULT
- .withSqlConformance(config.getParserConfig.conformance()))
+ .withConformance(config.getParserConfig.conformance()))
}
/**
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
index 43377e613df..897b6feff9b 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java
@@ -469,10 +469,10 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase {
.fromCase(BIGINT(), DEFAULT_NEGATIVE_BIGINT, 539222987)
.fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, 123)
.fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, -123)
- .fromCase(FLOAT(), 9234567891.12, 644633299)
+ .fromCase(FLOAT(), 9234567891.12f, 2147483647)
.fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, 123)
.fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, -123)
- .fromCase(DOUBLE(), 9234567891.12345, 644633299)
+ .fromCase(DOUBLE(), 9234567891.12345, 2147483647)
// Not supported - no fix
.failValidation(DATE(), DEFAULT_DATE)
.failValidation(TIME(), DEFAULT_TIME)
@@ -526,7 +526,7 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase {
.fromCase(BIGINT(), DEFAULT_NEGATIVE_BIGINT, DEFAULT_NEGATIVE_BIGINT)
.fromCase(FLOAT(), DEFAULT_POSITIVE_FLOAT, 123L)
.fromCase(FLOAT(), DEFAULT_NEGATIVE_FLOAT, -123L)
- .fromCase(FLOAT(), 9234567891.12, 9234567891L)
+ .fromCase(FLOAT(), 9234567891.12f, (long) 9234567891.12f)
.fromCase(DOUBLE(), DEFAULT_POSITIVE_DOUBLE, 123L)
.fromCase(DOUBLE(), DEFAULT_NEGATIVE_DOUBLE, -123L)
.fromCase(DOUBLE(), 9234567891.12345, 9234567891L)
@@ -1107,14 +1107,14 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase {
CastTestSpecBuilder.testCastTo(TINYINT())
.fromCase(TINYINT(), Byte.MIN_VALUE, Byte.MIN_VALUE)
.fromCase(TINYINT(), Byte.MAX_VALUE, Byte.MAX_VALUE)
- .fromCase(TINYINT(), Byte.MIN_VALUE - 1, Byte.MAX_VALUE)
- .fromCase(TINYINT(), Byte.MAX_VALUE + 1, Byte.MIN_VALUE)
+ .fromCase(TINYINT(), (byte) (Byte.MIN_VALUE - 1), Byte.MAX_VALUE)
+ .fromCase(TINYINT(), (byte) (Byte.MAX_VALUE + 1), Byte.MIN_VALUE)
.build(),
CastTestSpecBuilder.testCastTo(SMALLINT())
.fromCase(SMALLINT(), Short.MIN_VALUE, Short.MIN_VALUE)
.fromCase(SMALLINT(), Short.MAX_VALUE, Short.MAX_VALUE)
- .fromCase(SMALLINT(), Short.MIN_VALUE - 1, Short.MAX_VALUE)
- .fromCase(SMALLINT(), Short.MAX_VALUE + 1, Short.MIN_VALUE)
+ .fromCase(SMALLINT(), (short) (Short.MIN_VALUE - 1), Short.MAX_VALUE)
+ .fromCase(SMALLINT(), (short) (Short.MAX_VALUE + 1), Short.MIN_VALUE)
.build(),
CastTestSpecBuilder.testCastTo(INT())
.fromCase(INT(), Integer.MIN_VALUE, Integer.MIN_VALUE)
@@ -1132,22 +1132,28 @@ public class CastFunctionITCase extends BuiltInFunctionTestBase {
.fromCase(DOUBLE(), -1.7976931348623157E308d, Float.NEGATIVE_INFINITY)
.build(),
CastTestSpecBuilder.testCastTo(DECIMAL(38, 0))
- .fromCase(TINYINT(), Byte.MIN_VALUE - 1, new BigDecimal(Byte.MIN_VALUE - 1))
- .fromCase(TINYINT(), Byte.MAX_VALUE + 1, new BigDecimal(Byte.MAX_VALUE + 1))
+ .fromCase(
+ TINYINT(),
+ (byte) (Byte.MIN_VALUE - 1),
+ BigDecimal.valueOf(Byte.MAX_VALUE))
+ .fromCase(
+ TINYINT(),
+ (byte) (Byte.MAX_VALUE + 1),
+ BigDecimal.valueOf(Byte.MIN_VALUE))
.fromCase(
SMALLINT(),
- Short.MIN_VALUE - 1,
- new BigDecimal(Short.MIN_VALUE - 1))
+ (short) (Short.MIN_VALUE - 1),
+ BigDecimal.valueOf(Short.MAX_VALUE))
.fromCase(
SMALLINT(),
- Short.MAX_VALUE + 1,
- new BigDecimal(Short.MAX_VALUE + 1))
+ (short) (Short.MAX_VALUE + 1),
+ BigDecimal.valueOf(Short.MIN_VALUE))
.fromCase(
- INT(), Integer.MIN_VALUE - 1, new BigDecimal(Integer.MIN_VALUE - 1))
+ INT(), Integer.MIN_VALUE - 1, BigDecimal.valueOf(Integer.MAX_VALUE))
.fromCase(
- INT(), Integer.MAX_VALUE + 1, new BigDecimal(Integer.MAX_VALUE + 1))
- .fromCase(BIGINT(), Long.MIN_VALUE - 1, new BigDecimal(Long.MIN_VALUE - 1))
- .fromCase(BIGINT(), Long.MAX_VALUE + 1, new BigDecimal(Long.MAX_VALUE + 1))
+ INT(), Integer.MAX_VALUE + 1, BigDecimal.valueOf(Integer.MIN_VALUE))
+ .fromCase(BIGINT(), Long.MIN_VALUE - 1, BigDecimal.valueOf(Long.MAX_VALUE))
+ .fromCase(BIGINT(), Long.MAX_VALUE + 1, BigDecimal.valueOf(Long.MIN_VALUE))
.build(),
CastTestSpecBuilder.testCastTo(DECIMAL(38, 32))
.fromCase(FLOAT(), -Float.MAX_VALUE, null)
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/BroadcastJoinHintTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/BroadcastJoinHintTest.xml
index 7ecc331564f..cad55e9881b 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/BroadcastJoinHintTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/BroadcastJoinHintTest.xml
@@ -702,16 +702,16 @@ LogicalProject(EXPR$0=[+($0, $cor0.a1)])
<Resource name="optimized rel plan">
<![CDATA[
Calc(select=[a1, b1])
-+- HashJoin(joinType=[InnerJoin], where=[=(a1, EXPR$0)], select=[a1, b1, EXPR$0], build=[left])
++- HashJoin(joinType=[InnerJoin], where=[=(a1, a10)], select=[a1, b1, a10], build=[left])
:- Exchange(distribution=[hash[a1]])
: +- TableSourceScan(table=[[default_catalog, default_database, T1]], fields=[a1, b1])
- +- Exchange(distribution=[hash[EXPR$0]])
- +- Calc(select=[EXPR$0])
+ +- Exchange(distribution=[hash[a1]])
+ +- Calc(select=[a1])
+- HashAggregate(isMerge=[true], groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Exchange(distribution=[hash[EXPR$0, a1]])
+- LocalHashAggregate(groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Calc(select=[+(a2, a1) AS EXPR$0, a1])
- +- NestedLoopJoin(joinType=[InnerJoin], where=[=(+(a2, a1), a1)], select=[a2, a1], build=[right])
+ +- NestedLoopJoin(joinType=[InnerJoin], where=[=(a1, +(a2, a1))], select=[a2, a1], build=[right])
:- Calc(select=[a2])
: +- HashJoin(joinType=[InnerJoin], where=[=(a2, a3)], select=[a2, a3], isBroadcast=[true], build=[right])
: :- TableSourceScan(table=[[default_catalog, default_database, T2, project=[a2], metadata=[]]], fields=[a2], hints=[[[ALIAS options:[T2]]]])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/NestLoopJoinHintTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/NestLoopJoinHintTest.xml
index f771fa2f933..eadb478fbaa 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/NestLoopJoinHintTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/NestLoopJoinHintTest.xml
@@ -701,16 +701,16 @@ LogicalProject(EXPR$0=[+($0, $cor0.a1)])
<Resource name="optimized rel plan">
<![CDATA[
Calc(select=[a1, b1])
-+- HashJoin(joinType=[InnerJoin], where=[=(a1, EXPR$0)], select=[a1, b1, EXPR$0], build=[left])
++- HashJoin(joinType=[InnerJoin], where=[=(a1, a10)], select=[a1, b1, a10], build=[left])
:- Exchange(distribution=[hash[a1]])
: +- TableSourceScan(table=[[default_catalog, default_database, T1]], fields=[a1, b1])
- +- Exchange(distribution=[hash[EXPR$0]])
- +- Calc(select=[EXPR$0])
+ +- Exchange(distribution=[hash[a1]])
+ +- Calc(select=[a1])
+- HashAggregate(isMerge=[true], groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Exchange(distribution=[hash[EXPR$0, a1]])
+- LocalHashAggregate(groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Calc(select=[+(a2, a1) AS EXPR$0, a1])
- +- NestedLoopJoin(joinType=[InnerJoin], where=[=(+(a2, a1), a1)], select=[a2, a1], build=[right])
+ +- NestedLoopJoin(joinType=[InnerJoin], where=[=(a1, +(a2, a1))], select=[a2, a1], build=[right])
:- Calc(select=[a2])
: +- NestedLoopJoin(joinType=[InnerJoin], where=[=(a2, a3)], select=[a2, a3], build=[right])
: :- TableSourceScan(table=[[default_catalog, default_database, T2, project=[a2], metadata=[]]], fields=[a2], hints=[[[ALIAS options:[T2]]]])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleHashJoinHintTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleHashJoinHintTest.xml
index 00fbfb8ac69..4841125877d 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleHashJoinHintTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleHashJoinHintTest.xml
@@ -721,16 +721,16 @@ LogicalProject(EXPR$0=[+($0, $cor0.a1)])
<Resource name="optimized rel plan">
<![CDATA[
Calc(select=[a1, b1])
-+- HashJoin(joinType=[InnerJoin], where=[=(a1, EXPR$0)], select=[a1, b1, EXPR$0], build=[left])
++- HashJoin(joinType=[InnerJoin], where=[=(a1, a10)], select=[a1, b1, a10], build=[left])
:- Exchange(distribution=[hash[a1]])
: +- TableSourceScan(table=[[default_catalog, default_database, T1]], fields=[a1, b1])
- +- Exchange(distribution=[hash[EXPR$0]])
- +- Calc(select=[EXPR$0])
+ +- Exchange(distribution=[hash[a1]])
+ +- Calc(select=[a1])
+- HashAggregate(isMerge=[true], groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Exchange(distribution=[hash[EXPR$0, a1]])
+- LocalHashAggregate(groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Calc(select=[+(a2, a1) AS EXPR$0, a1])
- +- NestedLoopJoin(joinType=[InnerJoin], where=[=(+(a2, a1), a1)], select=[a2, a1], build=[right])
+ +- NestedLoopJoin(joinType=[InnerJoin], where=[=(a1, +(a2, a1))], select=[a2, a1], build=[right])
:- Calc(select=[a2])
: +- HashJoin(joinType=[InnerJoin], where=[=(a2, a3)], select=[a2, a3], build=[right])
: :- Exchange(distribution=[hash[a2]])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleMergeJoinHintTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleMergeJoinHintTest.xml
index 9600cb51a53..3888bcc893a 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleMergeJoinHintTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/hints/batch/ShuffleMergeJoinHintTest.xml
@@ -721,16 +721,16 @@ LogicalProject(EXPR$0=[+($0, $cor0.a1)])
<Resource name="optimized rel plan">
<![CDATA[
Calc(select=[a1, b1])
-+- HashJoin(joinType=[InnerJoin], where=[=(a1, EXPR$0)], select=[a1, b1, EXPR$0], build=[left])
++- HashJoin(joinType=[InnerJoin], where=[=(a1, a10)], select=[a1, b1, a10], build=[left])
:- Exchange(distribution=[hash[a1]])
: +- TableSourceScan(table=[[default_catalog, default_database, T1]], fields=[a1, b1])
- +- Exchange(distribution=[hash[EXPR$0]])
- +- Calc(select=[EXPR$0])
+ +- Exchange(distribution=[hash[a1]])
+ +- Calc(select=[a1])
+- HashAggregate(isMerge=[true], groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Exchange(distribution=[hash[EXPR$0, a1]])
+- LocalHashAggregate(groupBy=[EXPR$0, a1], select=[EXPR$0, a1])
+- Calc(select=[+(a2, a1) AS EXPR$0, a1])
- +- NestedLoopJoin(joinType=[InnerJoin], where=[=(+(a2, a1), a1)], select=[a2, a1], build=[right])
+ +- NestedLoopJoin(joinType=[InnerJoin], where=[=(a1, +(a2, a1))], select=[a2, a1], build=[right])
:- Calc(select=[a2])
: +- SortMergeJoin(joinType=[InnerJoin], where=[=(a2, a3)], select=[a2, a3])
: :- Exchange(distribution=[hash[a2]])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml
index 892c58622a2..8160f58a149 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml
@@ -96,7 +96,7 @@ FlinkLogicalCalc(select=[name, id, amount, price], where=[SEARCH(name, Sarg[_UTF
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
-+- LogicalFilter(condition=[>($3, 10.1)])
++- LogicalFilter(condition=[>($3, CAST(10.1:DECIMAL(3, 1)):INTEGER NOT NULL)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
@@ -150,7 +150,7 @@ FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, MyTable,
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
-+- LogicalFilter(condition=[>($2, 1.1)])
++- LogicalFilter(condition=[>($2, CAST(1.1:DECIMAL(2, 1)):INTEGER NOT NULL)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRuleTest.xml
index b343e3798f7..b40aebe273c 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRuleTest.xml
@@ -101,7 +101,7 @@ LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
-+- LogicalFilter(condition=[>($3, 10.1)])
++- LogicalFilter(condition=[>($3, CAST(10.1:DECIMAL(3, 1)):INTEGER NOT NULL)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [filterPushedDown=[false], filter=[]]]])
]]>
</Resource>
@@ -159,7 +159,7 @@ LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
-+- LogicalFilter(condition=[>($2, 1.1)])
++- LogicalFilter(condition=[>($2, CAST(1.1:DECIMAL(2, 1)):INTEGER NOT NULL)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [filterPushedDown=[false], filter=[]]]])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml
index 660e96d9c66..e396f27ab55 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml
@@ -101,7 +101,7 @@ LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
-+- LogicalFilter(condition=[>($3, 10.1)])
++- LogicalFilter(condition=[>($3, CAST(10.1:DECIMAL(3, 1)):INTEGER NOT NULL)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
@@ -159,7 +159,7 @@ LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
<Resource name="ast">
<![CDATA[
LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
-+- LogicalFilter(condition=[>($2, 1.1)])
++- LogicalFilter(condition=[>($2, CAST(1.1:DECIMAL(2, 1)):INTEGER NOT NULL)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml
index 0775ff1a311..6a8db57acf8 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml
@@ -73,7 +73,7 @@ LogicalAggregate(group=[{}], EXPR$0=[MIN($0)])
})], max_t2h=[$SCALAR_QUERY({
LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
LogicalProject(t2h=[$7])
- LogicalFilter(condition=[=($0, $cor1.t1a)])
+ LogicalFilter(condition=[=($0, $cor0.t1a)])
LogicalTableScan(table=[[default_catalog, default_database, t2, source: [TestTableSource(t2a, t2b, t2c, t2d, t2e, t2f, t2g, t2h, t2i)]]])
})])
+- LogicalFilter(condition=[=($0, _UTF-16LE'test')])
@@ -99,4 +99,34 @@ LogicalProject(min_t3d=[$9], max_t2h=[$11])
]]>
</Resource>
</TestCase>
+ <TestCase name="testWithFilterJoinCorrelate">
+ <Resource name="sql">
+ <![CDATA[
+SELECT t1a
+FROM t1
+WHERE EXISTS (SELECT max(t2h) FROM t2
+ LEFT OUTER JOIN t1 ttt
+ ON t2.t2a=t1.t1a)
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(t1a=[$0])
++- LogicalFilter(condition=[EXISTS({
+LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+ LogicalProject(t2h=[$7])
+ LogicalJoin(condition=[=($0, $cor0.t1a)], joinType=[left])
+ LogicalTableScan(table=[[default_catalog, default_database, t2, source: [TestTableSource(t2a, t2b, t2c, t2d, t2e, t2f, t2g, t2h, t2i)]]])
+ LogicalTableScan(table=[[default_catalog, default_database, t1, source: [TestTableSource(t1a, t1b, t1c, t1d, t1e, t1f, t1g, t1h, t1i)]]])
+})], variablesSet=[[$cor0]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, t1, source: [TestTableSource(t1a, t1b, t1c, t1d, t1e, t1f, t1g, t1h, t1i)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(t1a=[$0])
++- LogicalTableScan(table=[[default_catalog, default_database, t1, source: [TestTableSource(t1a, t1b, t1c, t1d, t1e, t1f, t1g, t1h, t1i)]]])
+]]>
+ </Resource>
+ </TestCase>
</Root>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/CalcTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/CalcTest.xml
index f27489805c3..f68e26af411 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/CalcTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/CalcTest.xml
@@ -107,14 +107,14 @@ Calc(select=[*org.apache.flink.table.planner.expressions.utils.Func24$$2da7dcb3c
<TestCase name="testNotIn">
<Resource name="ast">
<![CDATA[
-LogicalFilter(condition=[OR(SEARCH($1, Sarg[(-∞..1), (1..2), (2..3), (3..4), (4..5), (5..6), (6..7), (7..8), (8..9), (9..10), (10..11), (11..12), (12..13), (13..14), (14..15), (15..16), (16..17), (17..18), (18..19), (19..20), (20..21), (21..22), (22..23), (23..24), (24..25), (25..26), (26..27), (27..28), (28..29), (29..30), (30..+∞)]), SEARCH($2, Sarg[(-∞.._UTF-16LE'xx'), (_UTF-16LE'xx'..+∞)]:CHAR(2) CHARACTER SET "UTF-16LE"))])
+LogicalFilter(condition=[OR(SEARCH($1, Sarg[(-∞..1), (1..2), (2..3), (3..4), (4..5), (5..6), (6..7), (7..8), (8..9), (9..10), (10..11), (11..12), (12..13), (13..14), (14..15), (15..16), (16..17), (17..18), (18..19), (19..20), (20..21), (21..22), (22..23), (23..24), (24..25), (25..26), (26..27), (27..28), (28..29), (29..30), (30..+∞)]), <>($2, _UTF-16LE'xx'))])
+- LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[a, b, c], where=[(SEARCH(b, Sarg[(-∞..1), (1..2), (2..3), (3..4), (4..5), (5..6), (6..7), (7..8), (8..9), (9..10), (10..11), (11..12), (12..13), (13..14), (14..15), (15..16), (16..17), (17..18), (18..19), (19..20), (20..21), (21..22), (22..23), (23..24), (24..25), (25..26), (26..27), (27..28), (28..29), (29..30), (30..+∞)]) OR SEARCH(c, Sarg[(-∞.._UTF-16LE'xx'), (_UTF-16LE'xx'..+∞)]))])
+Calc(select=[a, b, c], where=[(SEARCH(b, Sarg[(-∞..1), (1..2), (2..3), (3..4), (4..5), (5..6), (6..7), (7..8), (8..9), (9..10), (10..11), (11..12), (12..13), (13..14), (14..15), (15..16), (16..17), (17..18), (18..19), (19..20), (20..21), (21..22), (22..23), (23..24), (24..25), (25..26), (26..27), (27..28), (28..29), (29..30), (30..+∞)]) OR (c <> 'xx'))])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.scala
index 9066f9373eb..bde051543f4 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.scala
@@ -96,7 +96,7 @@ class SubqueryCorrelateVariablesValidationTest extends SubQueryTestBase {
util.verifyRelPlan(sqlQuery)
}
- @Test(expected = classOf[TableException])
+ @Test
def testWithFilterJoinCorrelate(): Unit = {
val sqlQuery =
"""
diff --git a/flink-table/pom.xml b/flink-table/pom.xml
index 6ff401735b7..f7ae2e6be4e 100644
--- a/flink-table/pom.xml
+++ b/flink-table/pom.xml
@@ -78,12 +78,12 @@ under the License.
</dependencyManagement>
<properties>
- <calcite.version>1.30.0</calcite.version>
- <!-- Calcite 1.30.0 depends on 3.1.6,
+ <calcite.version>1.31.0</calcite.version>
+ <!-- Calcite 1.31.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.7.0</jsonpath.version>
- <guava.version>29.0-jre</guava.version>
+ <guava.version>31.1-jre</guava.version>
</properties>
</project>