From 6650bc89e66820688b8507d870762aa5f3e016b3 Mon Sep 17 00:00:00 2001
From: Sergey Nuyanzin <snuyanzin@gmail.com>
Date: Sun, 12 Mar 2023 10:50:38 +0100
Subject: [PATCH] [FLINK-31836][table] Upgrade Calcite version to 1.34.0

---
 flink-table/flink-sql-parser/pom.xml          |   4 +-
 .../flink-table-calcite-bridge/pom.xml        |   6 +-
 .../calcite/sql/type/SqlTypeFactoryImpl.java  |  12 +-
 .../sql/validate/SqlValidatorImpl.java        | 253 +++++++++++++-----
 .../calcite/sql2rel/RelDecorrelator.java      |  14 +-
 .../calcite/sql2rel/SqlToRelConverter.java    | 176 ++++++++++--
 .../org/apache/calcite/tools/RelBuilder.java  |   2 +-
 .../src/main/resources/META-INF/NOTICE        |   4 +-
 flink-table/pom.xml                           |   4 +-
 9 files changed, 360 insertions(+), 115 deletions(-)

diff --git a/flink-table/flink-sql-parser/pom.xml b/flink-table/flink-sql-parser/pom.xml
index ecfd5973557858..9a4cbf27bfaede 100644
--- a/flink-table/flink-sql-parser/pom.xml
+++ b/flink-table/flink-sql-parser/pom.xml
@@ -62,9 +62,9 @@ under the License.
 			<version>${calcite.version}</version>
 			<exclusions>
 				<!--
-				"mvn dependency:tree" as of Calcite 1.33.0:
+				"mvn dependency:tree" as of Calcite 1.34.0:
 
-				[INFO] +- org.apache.calcite:calcite-core:jar:1.33.0:compile
+				[INFO] +- org.apache.calcite:calcite-core:jar:1.34.0:compile
 				[INFO] |  +- org.apache.calcite.avatica:avatica-core:jar:1.23.0:compile
 				[INFO] |  +- org.apiguardian:apiguardian-api:jar:1.1.2:compile
 				[INFO] |  +- org.checkerframework:checker-qual:jar:3.12.0:compile
diff --git a/flink-table/flink-table-calcite-bridge/pom.xml b/flink-table/flink-table-calcite-bridge/pom.xml
index 4427bb28c9d2a8..3e3323f7379831 100644
--- a/flink-table/flink-table-calcite-bridge/pom.xml
+++ b/flink-table/flink-table-calcite-bridge/pom.xml
@@ -45,9 +45,9 @@ under the License.
 			<version>${calcite.version}</version>
 			<exclusions>
 				<!--
-				"mvn dependency:tree" as of Calcite 1.33.0:
-				[INFO] +- org.apache.calcite:calcite-core:jar:1.33.0:compile
-				[INFO] |  +- org.apache.calcite:calcite-linq4j:jar:1.33.0:compile
+				"mvn dependency:tree" as of Calcite 1.34.0:
+				[INFO] +- org.apache.calcite:calcite-core:jar:1.34.0:compile
+				[INFO] |  +- org.apache.calcite:calcite-linq4j:jar:1.34.0:compile
 				[INFO] |  +- org.locationtech.jts:jts-core:jar:1.19.0:compile
 				[INFO] |  +- com.fasterxml.jackson.core:jackson-annotations:jar:2.15.3:compile
 				[INFO] |  +- org.apache.calcite.avatica:avatica-core:jar:1.23.0:compile
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java
index 23f628ca1b2079..f88e621f684f73 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/type/SqlTypeFactoryImpl.java
@@ -476,9 +476,17 @@ private static void assertBasic(SqlTypeName typeName) {
                                 resultType, nullCount > 0 || nullableCount > 0);
                     }
                 }
+
+                if (type.getSqlTypeName() == resultType.getSqlTypeName()
+                        && type.getSqlTypeName().allowsPrec()
+                        && type.getPrecision() != resultType.getPrecision()) {
+                    final int precision =
+                            SqlTypeUtil.maxPrecision(
+                                    resultType.getPrecision(), type.getPrecision());
+
+                    resultType = createSqlType(type.getSqlTypeName(), precision);
+                }
             } else {
-                // TODO:  datetime precision details; for now we let
-                // leastRestrictiveByCast handle it
                 return null;
             }
         }
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 f6ee0093929bef..30a62a2806ed9f 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
@@ -497,9 +497,12 @@ private boolean expandSelectItem(
         inferUnknownTypes(targetType, 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),
+        // 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, in
+        //   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));
@@ -1457,6 +1460,7 @@ private void handleOffsetFetch(@Nullable SqlNode offset, @Nullable SqlNode fetch
                             null,
                             null,
                             null,
+                            null,
                             orderList,
                             orderBy.offset,
                             orderBy.fetch,
@@ -1481,6 +1485,7 @@ private void handleOffsetFetch(@Nullable SqlNode offset, @Nullable SqlNode fetch
                             null,
                             null,
                             null,
+                            null,
                             null);
                 }
 
@@ -1588,6 +1593,7 @@ private static void rewriteMerge(SqlMerge call) {
                         null,
                         null,
                         null,
+                        null,
                         null);
         call.setSourceSelect(select);
 
@@ -1614,6 +1620,7 @@ private static void rewriteMerge(SqlMerge call) {
                             null,
                             null,
                             null,
+                            null,
                             null);
             insertCall.setSource(select);
         }
@@ -1676,6 +1683,7 @@ private SqlNode rewriteUpdateToMerge(SqlUpdate updateCall, SqlNode selfJoinSrcEx
                         null,
                         null,
                         null,
+                        null,
                         null);
         source = SqlValidatorUtil.addAlias(source, UPDATE_SRC_ALIAS);
         SqlMerge mergeCall =
@@ -1742,6 +1750,7 @@ protected SqlSelect createSourceSelectForUpdate(SqlUpdate call) {
                 null,
                 null,
                 null,
+                null,
                 null);
     }
 
@@ -1772,6 +1781,7 @@ protected SqlSelect createSourceSelectForDelete(SqlDelete call) {
                 null,
                 null,
                 null,
+                null,
                 null);
     }
 
@@ -2783,6 +2793,9 @@ private void registerQuery(
                 clauseScopes.put(IdPair.of(select, Clause.WHERE), selectScope);
                 registerOperandSubQueries(selectScope, select, SqlSelect.WHERE_OPERAND);
 
+                // Register subqueries in the QUALIFY clause
+                registerOperandSubQueries(selectScope, select, SqlSelect.QUALIFY_OPERAND);
+
                 // Register FROM with the inherited scope 'parentScope', not
                 // 'selectScope', otherwise tables in the FROM clause would be
                 // able to see each other.
@@ -3597,12 +3610,11 @@ private RelDataType checkAndDeriveDataType(SqlIdentifier id, SqlNode node) {
         String name = id.names.get(0);
         SqlNameMatcher nameMatcher = getCatalogReader().nameMatcher();
         RelDataType rowType = getNamespaceOrThrow(node).getRowType();
-        RelDataType colType =
+        final RelDataTypeField field =
                 requireNonNull(
-                                nameMatcher.field(rowType, name),
-                                () -> "unable to find left field " + name + " in " + rowType)
-                        .getType();
-        return colType;
+                        nameMatcher.field(rowType, name),
+                        () -> "unable to find left field " + name + " in " + rowType);
+        return field.getType();
     }
 
     /**
@@ -3708,6 +3720,7 @@ protected void validateSelect(SqlSelect select, RelDataType targetRowType) {
         validateGroupClause(select);
         validateHavingClause(select);
         validateWindowClause(select);
+        validateQualifyClause(select);
         handleOffsetFetch(select.getOffset(), select.getFetch());
 
         // Validate the SELECT clause late, because a select item might
@@ -4156,6 +4169,34 @@ protected void validateWindowClause(SqlSelect select) {
         windowList.validate(this, windowScope);
     }
 
+    protected void validateQualifyClause(SqlSelect select) {
+        SqlNode qualifyNode = select.getQualify();
+        if (qualifyNode == null) {
+            return;
+        }
+
+        SqlValidatorScope qualifyScope = getSelectScope(select);
+
+        qualifyNode = extendedExpand(qualifyNode, qualifyScope, select, Clause.QUALIFY);
+        select.setQualify(qualifyNode);
+
+        inferUnknownTypes(booleanType, qualifyScope, qualifyNode);
+
+        qualifyNode.validate(this, qualifyScope);
+
+        final RelDataType type = deriveType(qualifyScope, qualifyNode);
+        if (!SqlTypeUtil.inBooleanFamily(type)) {
+            throw newValidationError(qualifyNode, RESOURCE.condMustBeBoolean("QUALIFY"));
+        }
+
+        boolean qualifyContainsWindowFunction = overFinder.findAgg(qualifyNode) != null;
+        if (!qualifyContainsWindowFunction) {
+            throw newValidationError(
+                    qualifyNode,
+                    RESOURCE.qualifyExpressionMustContainWindowFunction(qualifyNode.toString()));
+        }
+    }
+
     @Override
     public void validateWith(SqlWith with, SqlValidatorScope scope) {
         final SqlValidatorNamespace namespace = getNamespaceOrThrow(with);
@@ -4347,7 +4388,7 @@ protected void validateGroupClause(SqlSelect select) {
         // expand the expression in group list.
         List<SqlNode> expandedList = new ArrayList<>();
         for (SqlNode groupItem : groupList) {
-            SqlNode expandedItem = expandGroupByOrHavingExpr(groupItem, groupScope, select, false);
+            SqlNode expandedItem = extendedExpand(groupItem, groupScope, select, Clause.GROUP_BY);
             expandedList.add(expandedItem);
         }
         groupList = new SqlNodeList(expandedList, groupList.getParserPosition());
@@ -4473,7 +4514,7 @@ protected void validateHavingClause(SqlSelect select) {
         }
         final AggregatingScope havingScope = (AggregatingScope) getSelectScope(select);
         if (config.conformance().isHavingAlias()) {
-            SqlNode newExpr = expandGroupByOrHavingExpr(having, havingScope, select, true);
+            SqlNode newExpr = extendedExpand(having, havingScope, select, Clause.HAVING);
             if (having != newExpr) {
                 having = newExpr;
                 select.setHaving(newExpr);
@@ -4979,14 +5020,13 @@ protected void checkTypeAssignment(
         // matched
         boolean isUpdateModifiableViewTable = false;
         if (query instanceof SqlUpdate) {
-            final SqlNodeList targetColumnList = ((SqlUpdate) query).getTargetColumnList();
-            if (targetColumnList != null) {
-                final int targetColumnCnt = targetColumnList.size();
-                targetRowType =
-                        SqlTypeUtil.extractLastNFields(typeFactory, targetRowType, targetColumnCnt);
-                sourceRowType =
-                        SqlTypeUtil.extractLastNFields(typeFactory, sourceRowType, targetColumnCnt);
-            }
+            final SqlNodeList targetColumnList =
+                    requireNonNull(((SqlUpdate) query).getTargetColumnList());
+            final int targetColumnCount = targetColumnList.size();
+            targetRowType =
+                    SqlTypeUtil.extractLastNFields(typeFactory, targetRowType, targetColumnCount);
+            sourceRowType =
+                    SqlTypeUtil.extractLastNFields(typeFactory, sourceRowType, targetColumnCount);
             isUpdateModifiableViewTable = table.unwrap(ModifiableViewTable.class) != null;
         }
         if (SqlTypeUtil.equalAsStructSansNullability(
@@ -5491,13 +5531,12 @@ public void validateMatchRecognize(SqlCall call) {
                 node.validate(this, scope);
                 SqlIdentifier identifier;
                 if (node instanceof SqlBasicCall) {
-                    identifier = (SqlIdentifier) ((SqlBasicCall) node).operand(0);
+                    identifier = ((SqlBasicCall) node).operand(0);
                 } else {
                     identifier =
-                            (SqlIdentifier)
-                                    requireNonNull(
-                                            node,
-                                            () -> "order by field is null. All fields: " + orderBy);
+                            requireNonNull(
+                                    (SqlIdentifier) node,
+                                    () -> "order by field is null. All fields: " + orderBy);
                 }
 
                 if (allRows) {
@@ -5719,7 +5758,7 @@ private static String alias(SqlNode item) {
 
     public void validatePivot(SqlPivot pivot) {
         final PivotScope scope =
-                (PivotScope) requireNonNull(getJoinScope(pivot), () -> "joinScope for " + pivot);
+                requireNonNull((PivotScope) getJoinScope(pivot), () -> "joinScope for " + pivot);
 
         final PivotNamespace ns = getNamespaceOrThrow(pivot).unwrap(PivotNamespace.class);
         assert ns.rowType == null;
@@ -6144,9 +6183,10 @@ public SqlNode expand(SqlNode expr, SqlValidatorScope scope) {
         return newExpr;
     }
 
-    public SqlNode expandGroupByOrHavingExpr(
-            SqlNode expr, SqlValidatorScope scope, SqlSelect select, boolean havingExpression) {
-        final Expander expander = new ExtendedExpander(this, scope, select, expr, havingExpression);
+    /** Expands an expression in a GROUP BY, HAVING or QUALIFY clause. */
+    private SqlNode extendedExpand(
+            SqlNode expr, SqlValidatorScope scope, SqlSelect select, Clause clause) {
+        final Expander expander = new ExtendedExpander(this, scope, select, expr, clause);
         SqlNode newExpr = expander.go(expr);
         if (expr != newExpr) {
             setOriginal(newExpr, expr);
@@ -6154,6 +6194,10 @@ public SqlNode expandGroupByOrHavingExpr(
         return newExpr;
     }
 
+    public SqlNode extendedExpandGroupBy(SqlNode expr, SqlValidatorScope scope, SqlSelect select) {
+        return extendedExpand(expr, scope, select, Clause.GROUP_BY);
+    }
+
     @Override
     public boolean isSystemField(RelDataTypeField field) {
         return false;
@@ -6763,67 +6807,74 @@ static class SelectExpander extends Expander {
     static class ExtendedExpander extends Expander {
         final SqlSelect select;
         final SqlNode root;
-        final boolean havingExpr;
+        final Clause clause;
 
         ExtendedExpander(
                 SqlValidatorImpl validator,
                 SqlValidatorScope scope,
                 SqlSelect select,
                 SqlNode root,
-                boolean havingExpr) {
+                Clause clause) {
             super(validator, scope);
             this.select = select;
             this.root = root;
-            this.havingExpr = havingExpr;
+            this.clause = clause;
         }
 
         @Override
         public @Nullable SqlNode visit(SqlIdentifier id) {
-            if (id.isSimple()
-                    && (havingExpr
-                            ? validator.config().conformance().isHavingAlias()
-                            : validator.config().conformance().isGroupByAlias())) {
-                String name = id.getSimple();
-                SqlNode expr = null;
-                final SqlNameMatcher nameMatcher = validator.catalogReader.nameMatcher();
-                int n = 0;
-                for (SqlNode s : SqlNonNullableAccessors.getSelectList(select)) {
-                    final @Nullable String alias = SqlValidatorUtil.alias(s);
-                    if (alias != null && nameMatcher.matches(alias, name)) {
-                        expr = s;
-                        n++;
-                    }
-                }
-                if (n == 0) {
-                    return super.visit(id);
-                } else if (n > 1) {
-                    // More than one column has this alias.
-                    throw validator.newValidationError(id, RESOURCE.columnAmbiguous(name));
-                }
-                if (havingExpr && validator.isAggregate(root)) {
-                    return super.visit(id);
-                }
-                expr = stripAs(expr);
-                if (expr instanceof SqlIdentifier) {
-                    SqlIdentifier sid = (SqlIdentifier) expr;
-                    final SqlIdentifier fqId = getScope().fullyQualify(sid).identifier;
-                    expr = expandDynamicStar(sid, fqId);
-                }
-                return expr;
+            if (!id.isSimple()) {
+                return super.visit(id);
             }
-            if (id.isSimple()) {
+
+            final boolean replaceAliases = clause.shouldReplaceAliases(validator.config);
+            if (!replaceAliases) {
                 final SelectScope scope = validator.getRawSelectScope(select);
                 SqlNode node = expandCommonColumn(select, id, scope, validator);
                 if (node != id) {
                     return node;
                 }
+                return super.visit(id);
+            }
+
+            String name = id.getSimple();
+            SqlNode expr = null;
+            final SqlNameMatcher nameMatcher = validator.catalogReader.nameMatcher();
+            int n = 0;
+            for (SqlNode s : SqlNonNullableAccessors.getSelectList(select)) {
+                final @Nullable String alias = SqlValidatorUtil.alias(s);
+                if (alias != null && nameMatcher.matches(alias, name)) {
+                    expr = s;
+                    n++;
+                }
+            }
+
+            if (n == 0) {
+                return super.visit(id);
+            } else if (n > 1) {
+                // More than one column has this alias.
+                throw validator.newValidationError(id, RESOURCE.columnAmbiguous(name));
+            }
+            Iterable<SqlCall> allAggList = validator.aggFinder.findAll(ImmutableList.of(root));
+            for (SqlCall agg : allAggList) {
+                if (clause == Clause.HAVING && containsIdentifier(agg, id)) {
+                    return super.visit(id);
+                }
             }
-            return super.visit(id);
+
+            expr = stripAs(expr);
+            if (expr instanceof SqlIdentifier) {
+                SqlIdentifier sid = (SqlIdentifier) expr;
+                final SqlIdentifier fqId = getScope().fullyQualify(sid).identifier;
+                expr = expandDynamicStar(sid, fqId);
+            }
+
+            return expr;
         }
 
         @Override
         public @Nullable SqlNode visit(SqlLiteral literal) {
-            if (havingExpr || !validator.config().conformance().isGroupByOrdinal()) {
+            if (clause != Clause.GROUP_BY || !validator.config().conformance().isGroupByOrdinal()) {
                 return super.visit(literal);
             }
             boolean isOrdinalLiteral = literal == root;
@@ -6871,6 +6922,32 @@ static class ExtendedExpander extends Expander {
 
             return super.visit(literal);
         }
+
+        /**
+         * Returns whether a given node contains a {@link SqlIdentifier}.
+         *
+         * @param sqlNode a SqlNode
+         * @param target a SqlIdentifier
+         */
+        private boolean containsIdentifier(SqlNode sqlNode, SqlIdentifier target) {
+            try {
+                SqlVisitor<Void> visitor =
+                        new SqlBasicVisitor<Void>() {
+                            @Override
+                            public Void visit(SqlIdentifier identifier) {
+                                if (identifier.equalsDeep(target, Litmus.IGNORE)) {
+                                    throw new Util.FoundOne(target);
+                                }
+                                return super.visit(identifier);
+                            }
+                        };
+                sqlNode.accept(visitor);
+                return false;
+            } catch (Util.FoundOne e) {
+                Util.swallow(e, null);
+                return true;
+            }
+        }
     }
 
     /** Information about an identifier in a particular scope. */
@@ -7343,6 +7420,52 @@ private enum Clause {
         GROUP_BY,
         SELECT,
         ORDER,
-        CURSOR
+        CURSOR,
+        HAVING,
+        QUALIFY;
+
+        /**
+         * Determines if the extender should replace aliases with expanded values. For example:
+         *
+         * <blockquote>
+         *
+         * <pre>{@code
+         * SELECT a + a as twoA
+         * GROUP BY twoA
+         * }</pre>
+         *
+         * </blockquote>
+         *
+         * <p>turns into
+         *
+         * <blockquote>
+         *
+         * <pre>{@code
+         * SELECT a + a as twoA
+         * GROUP BY a + a
+         * }</pre>
+         *
+         * </blockquote>
+         *
+         * <p>This is determined both by the clause and the config.
+         *
+         * @param config The configuration
+         * @return Whether we should replace the alias with its expanded value
+         */
+        boolean shouldReplaceAliases(Config config) {
+            switch (this) {
+                case GROUP_BY:
+                    return config.conformance().isGroupByAlias();
+
+                case HAVING:
+                    return config.conformance().isHavingAlias();
+
+                case QUALIFY:
+                    return true;
+
+                default:
+                    throw Util.unexpected(this);
+            }
+        }
     }
 }
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 92bc6526eb6404..abb327f7361bed 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
@@ -1065,17 +1065,11 @@ private Frame decorrelateInputWithValueGenerator(RelNode rel, Frame frame) {
 
         // can directly add positions into corDefOutputs since join
         // does not change the output ordering from the inputs.
-        RelNode valueGen =
-                requireNonNull(
-                        createValueGenerator(corVarList, leftInputOutputCount, corDefOutputs),
-                        "createValueGenerator(...) is null");
+        final RelNode valueGen =
+                createValueGenerator(corVarList, leftInputOutputCount, corDefOutputs);
+        requireNonNull(valueGen, "valueGen");
 
-        RelNode join =
-                relBuilder
-                        .push(frame.r)
-                        .push(valueGen)
-                        .join(JoinRelType.INNER, relBuilder.literal(true), ImmutableSet.of())
-                        .build();
+        RelNode join = relBuilder.push(frame.r).push(valueGen).join(JoinRelType.INNER).build();
 
         // Join or Filter does not change the old input ordering. All
         // input fields from newLeftInput (i.e. the original input to the old
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 97da4ebe108a97..55e4172dd55a7e 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
@@ -98,6 +98,7 @@
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexOver;
 import org.apache.calcite.rex.RexPatternFieldRef;
 import org.apache.calcite.rex.RexRangeRef;
 import org.apache.calcite.rex.RexShuttle;
@@ -225,6 +226,7 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
 import static org.apache.calcite.linq4j.Nullness.castNonNull;
+import static org.apache.calcite.runtime.FlatLists.append;
 import static org.apache.calcite.sql.SqlUtil.stripAs;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -780,6 +782,8 @@ protected void convertSelectImpl(final Blackboard bb, SqlSelect select) {
             convertSelectList(bb, select, orderExprList);
         }
 
+        convertQualify(bb, select.getQualify());
+
         if (select.isDistinct()) {
             distinctify(bb, true);
         }
@@ -894,8 +898,7 @@ private void distinctify(Blackboard bb, boolean checkForDupExprs) {
         }
 
         assert rel != null : "rel must not be null, root = " + bb.root;
-        // Usual case: all of the expressions in the SELECT clause are
-        // different.
+        // Usual case: all expressions in the SELECT clause are different.
         final ImmutableBitSet groupSet = ImmutableBitSet.range(rel.getRowType().getFieldCount());
         rel = createAggregate(bb, groupSet, ImmutableList.of(groupSet), ImmutableList.of());
 
@@ -3368,8 +3371,8 @@ private void convertJoin(Blackboard bb, SqlJoin join) {
                         condition,
                         convertJoinType(join.getJoinType()));
         relBuilder.push(joinRel);
-        final RelNode newProjectRel = relBuilder.project(relBuilder.fields()).build();
-        bb.setRoot(newProjectRel, false);
+        relBuilder.project(relBuilder.fields());
+        bb.setRoot(relBuilder.build(), false);
     }
 
     private RexNode convertNaturalCondition(
@@ -3387,8 +3390,8 @@ private RexNode convertUsingCondition(
             SqlValidatorNamespace leftNamespace,
             SqlValidatorNamespace rightNamespace) {
         final SqlNodeList list =
-                (SqlNodeList)
-                        requireNonNull(join.getCondition(), () -> "getCondition for join " + join);
+                requireNonNull(
+                        (SqlNodeList) join.getCondition(), () -> "getCondition for join " + join);
         return convertUsing(
                 leftNamespace,
                 rightNamespace,
@@ -4316,10 +4319,9 @@ protected void collectInsertTargets(
 
     private RelNode convertDelete(SqlDelete call) {
         RelOptTable targetTable = getTargetTable(call);
-        RelNode sourceRel =
-                convertSelect(
-                        requireNonNull(call.getSourceSelect(), () -> "sourceSelect for " + call),
-                        false);
+        final SqlSelect sourceSelect =
+                requireNonNull(call.getSourceSelect(), () -> "sourceSelect for " + call);
+        RelNode sourceRel = convertSelect(sourceSelect, false);
         return LogicalTableModify.create(
                 targetTable,
                 catalogReader,
@@ -4331,11 +4333,9 @@ private RelNode convertDelete(SqlDelete call) {
     }
 
     private RelNode convertUpdate(SqlUpdate call) {
-        final SqlValidatorScope scope =
-                validator()
-                        .getWhereScope(
-                                requireNonNull(
-                                        call.getSourceSelect(), () -> "sourceSelect for " + call));
+        final SqlSelect sourceSelect =
+                requireNonNull(call.getSourceSelect(), () -> "sourceSelect for " + call);
+        final SqlValidatorScope scope = validator().getWhereScope(sourceSelect);
         Blackboard bb = createBlackboard(scope, null, false);
 
         replaceSubQueries(bb, call, RelOptUtil.Logic.TRUE_FALSE_UNKNOWN);
@@ -4350,14 +4350,11 @@ private RelNode convertUpdate(SqlUpdate call) {
             RelDataTypeField field =
                     SqlValidatorUtil.getTargetField(
                             targetRowType, typeFactory, id, catalogReader, targetTable);
-            assert field != null : "column " + id.toString() + " not found";
+            assert field != null : "column " + id + " not found";
             targetColumnNameList.add(field.getName());
         }
 
-        RelNode sourceRel =
-                convertSelect(
-                        requireNonNull(call.getSourceSelect(), () -> "sourceSelect for " + call),
-                        false);
+        RelNode sourceRel = convertSelect(sourceSelect, false);
 
         bb.setRoot(sourceRel, false);
         ImmutableList.Builder<RexNode> rexNodeSourceExpressionListBuilder = ImmutableList.builder();
@@ -4403,10 +4400,9 @@ private RelNode convertMerge(SqlMerge call) {
 
         // first, convert the merge's source select to construct the columns
         // from the target table and the set expressions in the update call
-        RelNode mergeSourceRel =
-                convertSelect(
-                        requireNonNull(call.getSourceSelect(), () -> "sourceSelect for " + call),
-                        false);
+        final SqlSelect sourceSelect =
+                requireNonNull(call.getSourceSelect(), () -> "sourceSelect for " + call);
+        RelNode mergeSourceRel = convertSelect(sourceSelect, false);
 
         // then, convert the insert statement so we can get the insert
         // values expressions
@@ -4683,6 +4679,10 @@ private void convertSelectList(Blackboard bb, SqlSelect select, List<SqlNode> or
         selectList = validator().expandStar(selectList, select, false);
 
         replaceSubQueries(bb, selectList, RelOptUtil.Logic.TRUE_FALSE_UNKNOWN);
+        replaceSubQueries(
+                bb,
+                new SqlNodeList(orderList, SqlParserPos.ZERO),
+                RelOptUtil.Logic.TRUE_FALSE_UNKNOWN);
 
         List<String> fieldNames = new ArrayList<>();
         final List<RexNode> exprs = new ArrayList<>();
@@ -4779,6 +4779,118 @@ private static String deriveAlias(
         return alias;
     }
 
+    private void convertQualify(Blackboard bb, @Nullable SqlNode qualify) {
+        if (qualify == null) {
+            return;
+        }
+
+        final LogicalProject projectionFromSelect =
+                requireNonNull((LogicalProject) bb.root, "root");
+
+        // Convert qualify SqlNode to a RexNode
+        replaceSubQueries(bb, qualify, RelOptUtil.Logic.UNKNOWN_AS_FALSE);
+        final RelNode originalRoot = requireNonNull(bb.root, "root");
+        RexNode qualifyRexNode;
+        try {
+            // Set the root to the input of the project,
+            // since QUALIFY might have an expression in the OVER clause
+            // that references a column not in the SELECT.
+            bb.setRoot(projectionFromSelect.getInput(), false);
+            qualifyRexNode = bb.convertExpression(qualify);
+        } finally {
+            bb.setRoot(originalRoot, false);
+        }
+
+        // Check to see if the qualify expression has a referenced expression and
+        // do some referencing accordingly
+        final RexNode qualifyWithReferencesRexNode =
+                qualifyRexNode.accept(new DuplicateEliminator(projectionFromSelect.getProjects()));
+
+        // Create a Project with the QUALIFY expression
+        if (qualifyWithReferencesRexNode.equals(qualifyRexNode)) {
+            // The QUALIFY expression does not depend on any references like so:
+            //
+            //  SELECT A, B
+            //  FROM tbl
+            //  QUALIFY WINDOW(C) = 1
+            //
+            // Meaning we should generate a plan like:
+            //  Project(A, B, WINDOW(C) = 1 as QualifyExpression)
+            //    TableScan(tbl)
+            //
+            relBuilder
+                    .push(projectionFromSelect.getInput())
+                    .project(
+                            append(projectionFromSelect.getProjects(), qualifyRexNode),
+                            append(
+                                    projectionFromSelect.getRowType().getFieldNames(),
+                                    "QualifyExpression"));
+        } else {
+            // The QUALIFY expression depended on a reference meaning
+            // we need to introduce an extra project like so:
+            //
+            //  SELECT A, B, WINDOW(C) as window_val
+            //  FROM tbl
+            //  QUALIFY window_val = 1
+            //
+            // Meaning we should generate a plan like:
+            //
+            //  Project($0, $1, $2, =($2, 1) as QualifyExpression)
+            //    Project(A, B, WINDOW(C) as window_val)
+            //      TableScan(tbl)
+            //
+            // This is a very specific application of Common Subexpression Elimination
+            // (CSE), since the window value pops up twice.
+            relBuilder
+                    .push(requireNonNull(bb.root, "root"))
+                    .project(
+                            append(relBuilder.fields(), qualifyWithReferencesRexNode),
+                            append(
+                                    relBuilder.peek().getRowType().getFieldNames(),
+                                    "QualifyExpression"));
+        }
+
+        // Filter on that extra column
+        relBuilder.filter(Util.last(relBuilder.fields()));
+
+        // Remove that extra column from the projection
+        relBuilder.project(
+                Util.first(relBuilder.fields(), projectionFromSelect.getProjects().size()));
+
+        // Update the root
+        bb.setRoot(relBuilder.build(), false);
+    }
+
+    /**
+     * Eliminates a common sub-expression by looking for a {@link RexNode} in the expressions of a
+     * {@link Project}; if found, returns a refIndex instead of the raw node.
+     */
+    private static final class DuplicateEliminator extends RexShuttle {
+        private final List<RexNode> projects;
+
+        DuplicateEliminator(List<RexNode> projects) {
+            this.projects = projects;
+        }
+
+        @Override
+        public RexNode visitCall(RexCall call) {
+            final int i = projects.indexOf(call);
+            if (i >= 0) {
+                return new RexInputRef(i, projects.get(i).getType());
+            }
+            return super.visitCall(call);
+        }
+
+        @Override
+        public RexNode visitOver(RexOver over) {
+            final int i = projects.indexOf(over);
+            if (i >= 0) {
+                return new RexInputRef(i, projects.get(i).getType());
+            }
+            return over;
+        }
+    }
+
     /** Converts a WITH sub-query into a relational expression. */
     public RelRoot convertWith(SqlWith with, boolean top) {
         return convertQuery(with.body, false, top);
@@ -6745,15 +6857,23 @@ default boolean isExplain() {
         Config withExplain(boolean explain);
 
         /**
-         * Returns the {@code expand} option. Controls whether to expand sub-queries. If false, each
-         * sub-query becomes a {@link org.apache.calcite.rex.RexSubQuery}.
+         * Returns the {@code expand} option. Controls whether to expand sub-queries. If false (the
+         * default), each sub-query becomes a {@link org.apache.calcite.rex.RexSubQuery}.
+         *
+         * <p>Setting {@code expand} to true is deprecated. Expansion still works, but there will be
+         * less development effort in that area.
          */
         @Value.Default
         default boolean isExpand() {
-            return true;
+            return false;
         }
 
-        /** Sets {@link #isExpand()}. */
+        /**
+         * Sets {@link #isExpand()}.
+         *
+         * <p>Expansion is deprecated. We recommend that you do not call this method, and use the
+         * default value of {@link #isExpand()}, false.
+         */
         Config withExpand(boolean expand);
 
         /**
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 0c21d0af78c47a..22beba442a8c22 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
@@ -2288,7 +2288,7 @@ public RelBuilder projectNamed(
             @Nullable Iterable<? extends @Nullable String> fieldNames,
             boolean force,
             Iterable<CorrelationId> variablesSet) {
-        @SuppressWarnings("unchecked")
+        @SuppressWarnings({"unchecked", "rawtypes"})
         final List<? extends RexNode> nodeList =
                 nodes instanceof List ? (List) nodes : ImmutableList.copyOf(nodes);
         final List<@Nullable String> fieldNameList =
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 9f70b412e31894..7475cc50b20479 100644
--- a/flink-table/flink-table-planner/src/main/resources/META-INF/NOTICE
+++ b/flink-table/flink-table-planner/src/main/resources/META-INF/NOTICE
@@ -8,8 +8,8 @@ This project bundles the following dependencies under the Apache Software Licens
 
 - com.google.guava:guava:32.1.3-jre
 - com.google.guava:failureaccess:1.0.1
-- org.apache.calcite:calcite-core:1.33.0
-- org.apache.calcite:calcite-linq4j:1.33.0
+- org.apache.calcite:calcite-core:1.34.0
+- org.apache.calcite:calcite-linq4j:1.34.0
 - org.apache.calcite.avatica:avatica-core:1.23.0
 - org.apache.commons:commons-math3:3.6.1
 - commons-codec:commons-codec:1.15
diff --git a/flink-table/pom.xml b/flink-table/pom.xml
index 65b77daaf2b621..4577006256b090 100644
--- a/flink-table/pom.xml
+++ b/flink-table/pom.xml
@@ -78,8 +78,8 @@ under the License.
 	</dependencyManagement>
 
 	<properties>
-		<calcite.version>1.33.0</calcite.version>
-		<!-- Calcite 1.33.0 depends on 3.1.8,
+		<calcite.version>1.34.0</calcite.version>
+		<!-- Calcite 1.34.0 depends on 3.1.8,
 		at the same time minimum 3.1.x Janino version passing Flink tests without WAs is 3.1.10,
 		more details are in FLINK-27995 -->
 		<janino.version>3.1.10</janino.version>