You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2024/09/25 10:18:28 UTC
(flink) 02/05: [FLINK-36273][table] Remove deprecated Table/SQL configuration related FLINK-35473 in 2.0
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e60278d768459ed73e27a970f6a6c95c3474d85c
Author: xuyang <xy...@163.com>
AuthorDate: Sat Sep 14 14:20:09 2024 +0800
[FLINK-36273][table] Remove deprecated Table/SQL configuration related FLINK-35473 in 2.0
---
.../table/client/config/SqlClientOptions.java | 13 --
.../table/client/gateway/ResultDescriptor.java | 4 +-
.../table/client/cli/CliTableauResultViewTest.java | 6 +-
.../src/test/resources/sql/select.q | 16 +-
.../src/test/resources/sql/select_batch.q | 8 +-
.../src/test/resources/sql/table.q | 4 +-
.../table/api/config/OptimizerConfigOptions.java | 22 ---
.../table/planner/hint/LookupJoinHintOptions.java | 203 ---------------------
.../PushFilterIntoLegacyTableSourceScanRule.java | 12 --
.../logical/PushFilterIntoSourceScanRuleBase.java | 11 --
.../PushPartitionIntoTableSourceScanRule.java | 2 -
.../plan/rules/logical/SubQueryDecorrelator.java | 25 +--
.../batch/PushLocalAggIntoScanRuleBase.java | 10 -
.../batch/PushLocalHashAggIntoScanRule.java | 6 +-
.../PushLocalHashAggWithCalcIntoScanRule.java | 6 +-
.../batch/PushLocalSortAggIntoScanRule.java | 6 +-
.../PushLocalSortAggWithCalcIntoScanRule.java | 6 +-
...ushLocalSortAggWithSortAndCalcIntoScanRule.java | 6 +-
.../PushLocalSortAggWithSortIntoScanRule.java | 6 +-
.../codegen/agg/batch/HashAggCodeGenerator.scala | 83 ---------
.../plan/metadata/FlinkRelMdColumnNullCount.scala | 5 +-
.../plan/metadata/FlinkRelMdDistinctRowCount.scala | 5 +-
.../plan/metadata/FlinkRelMdDistribution.scala | 15 +-
.../planner/plan/metadata/FlinkRelMdRowCount.scala | 23 +--
.../plan/metadata/SelectivityEstimator.scala | 5 +-
.../plan/nodes/logical/FlinkLogicalSort.scala | 7 +-
.../batch/BatchPhysicalHashAggregate.scala | 7 +-
.../batch/BatchPhysicalOverAggregateBase.scala | 12 +-
.../batch/BatchPhysicalPythonGroupAggregate.scala | 7 +-
.../nodes/physical/batch/BatchPhysicalRank.scala | 7 +-
.../batch/BatchPhysicalSortAggregate.scala | 7 +-
.../table/planner/plan/optimize/RelNodeBlock.scala | 45 +----
.../rules/logical/JoinDeriveNullFilterRule.scala | 27 +--
...ushPartitionIntoLegacyTableSourceScanRule.scala | 4 +-
.../physical/batch/BatchPhysicalHashJoinRule.scala | 15 --
.../physical/batch/BatchPhysicalJoinRuleBase.scala | 44 +----
.../batch/BatchPhysicalSortMergeJoinRule.scala | 117 +++---------
.../physical/batch/BatchPhysicalSortRule.scala | 25 +--
.../physical/stream/IncrementalAggregateRule.scala | 27 ---
.../table/planner/plan/utils/FlinkRelOptUtil.scala | 8 +-
.../table/planner/plan/utils/FlinkRexUtil.scala | 46 +----
.../planner/plan/utils/IntervalJoinUtil.scala | 5 +-
.../flink/table/planner/plan/utils/RankUtil.scala | 8 +-
.../planner/plan/utils/RexNodeExtractor.scala | 23 +--
.../planner/plan/batch/sql/RemoveCollationTest.xml | 18 +-
.../planner/plan/batch/sql/RemoveShuffleTest.xml | 64 +++----
.../table/planner/plan/batch/sql/SortTest.xml | 9 +-
.../planner/plan/batch/sql/SubplanReuseTest.xml | 2 +-
.../batch/RemoveRedundantLocalHashAggRuleTest.xml | 36 ++--
.../plan/batch/sql/RemoveCollationTest.scala | 8 -
.../planner/plan/batch/sql/RemoveShuffleTest.scala | 28 ---
.../planner/plan/batch/sql/SortLimitTest.scala | 11 --
.../table/planner/plan/batch/sql/SortTest.scala | 6 -
.../planner/plan/batch/sql/SubplanReuseTest.scala | 6 -
.../planner/plan/common/JoinReorderTestBase.scala | 14 +-
.../logical/JoinDeriveNullFilterRuleTest.scala | 2 -
.../RemoveRedundantLocalHashAggRuleTest.scala | 6 -
.../planner/plan/utils/FlinkRexUtilTest.scala | 23 +--
.../planner/plan/utils/RexNodeExtractorTest.scala | 38 ++--
.../planner/runtime/batch/sql/CalcITCase.scala | 28 ++-
60 files changed, 216 insertions(+), 1032 deletions(-)
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java
index 488361b5881..733cedeb1d1 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/config/SqlClientOptions.java
@@ -22,7 +22,6 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.docs.Documentation;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
-import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.client.cli.parser.SyntaxHighlightStyle;
/** Options used in sql client. */
@@ -57,18 +56,6 @@ public class SqlClientOptions {
"Determine whether to output the verbose output to the console. If set the option true, it will print the exception stack. Otherwise, it only output the cause.");
// Display options
- /** @deprecated Please use {@link TableConfigOptions#DISPLAY_MAX_COLUMN_WIDTH} instead. */
- @Deprecated
- @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
- public static final ConfigOption<Integer> DISPLAY_MAX_COLUMN_WIDTH =
- ConfigOptions.key("sql-client.display.max-column-width")
- .intType()
- .defaultValue(30)
- .withFallbackKeys(TableConfigOptions.DISPLAY_MAX_COLUMN_WIDTH.key())
- .withDescription(
- "Deprecated, please use table.display.max-column-width instead. When printing the query results, this parameter determines the number of characters shown on screen before truncating. "
- + "This only applies to columns with variable-length types (e.g. CHAR, VARCHAR, STRING) in streaming mode. "
- + "Fixed-length types and all types in batch mode are printed using a deterministic column width.");
@Documentation.TableOption(execMode = Documentation.ExecMode.BATCH)
public static final ConfigOption<Boolean> DISPLAY_QUERY_TIME_COST =
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java
index d1c5c6dd66e..621206f45b9 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/ResultDescriptor.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.client.gateway;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.config.ResultMode;
import org.apache.flink.table.client.gateway.result.ChangelogCollectResult;
@@ -31,7 +32,6 @@ import org.apache.flink.table.utils.print.RowDataToStringConverter;
import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
import static org.apache.flink.table.client.config.ResultMode.CHANGELOG;
import static org.apache.flink.table.client.config.ResultMode.TABLE;
-import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_MAX_COLUMN_WIDTH;
import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_QUERY_TIME_COST;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_MAX_TABLE_RESULT_ROWS;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_RESULT_MODE;
@@ -92,7 +92,7 @@ public class ResultDescriptor {
}
public int maxColumnWidth() {
- return config.get(DISPLAY_MAX_COLUMN_WIDTH);
+ return config.get(TableConfigOptions.DISPLAY_MAX_COLUMN_WIDTH);
}
public boolean isPrintQueryTimeCost() {
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
index 3725e02ac43..eedb6ee2590 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.config.TableConfigOptions;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.config.ResultMode;
@@ -59,7 +60,6 @@ import java.util.stream.Collectors;
import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
-import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_MAX_COLUMN_WIDTH;
import static org.apache.flink.table.client.config.SqlClientOptions.DISPLAY_QUERY_TIME_COST;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_RESULT_MODE;
import static org.assertj.core.api.Assertions.assertThat;
@@ -229,7 +229,7 @@ class CliTableauResultViewTest {
view.close();
// adjust the max column width for printing
- testConfig.set(DISPLAY_MAX_COLUMN_WIDTH, 80);
+ testConfig.set(TableConfigOptions.DISPLAY_MAX_COLUMN_WIDTH, 80);
TestChangelogResult collectResult = createNewTestChangelogResult();
view =
@@ -456,7 +456,7 @@ class CliTableauResultViewTest {
+ System.lineSeparator());
// adjust the max column width for printing
- testConfig.set(DISPLAY_MAX_COLUMN_WIDTH, 80);
+ testConfig.set(TableConfigOptions.DISPLAY_MAX_COLUMN_WIDTH, 80);
collectResult = createNewTestChangelogResult();
view =
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/select.q b/flink-table/flink-sql-client/src/test/resources/sql/select.q
index f67853190f3..81657060f3a 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/select.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/select.q
@@ -116,7 +116,7 @@ Received a total of 2 rows
!ok
# ==========================================================================
-# Testing behavior of sql-client.display.max-column-width
+# Testing behavior of table.display.max-column-width
# Only variable width columns are impacted at the moment => STRING, but not TIMESTAMP nor BOOLEAN
# ==========================================================================
@@ -179,7 +179,7 @@ Received a total of 4 rows
# test original config option key
-SET 'sql-client.display.max-column-width' = '10';
+SET 'table.display.max-column-width' = '10';
[INFO] Execute statement succeeded.
!info
@@ -196,7 +196,7 @@ Received a total of 4 rows
!ok
-SET 'sql-client.display.max-column-width' = '40';
+SET 'table.display.max-column-width' = '40';
[INFO] Execute statement succeeded.
!info
@@ -217,7 +217,7 @@ DROP TEMPORARY VIEW testUserData;
[INFO] Execute statement succeeded.
!info
-SET 'sql-client.display.max-column-width' = '30';
+SET 'table.display.max-column-width' = '30';
[INFO] Execute statement succeeded.
!info
@@ -260,7 +260,7 @@ FROM (VALUES
!ok
# ==========================================================================
-# Testing behavior of sql-client.display.max-column-width
+# Testing behavior of table.display.max-column-width
# Only variable width columns are impacted at the moment => STRING, but not TIMESTAMP nor BOOLEAN
# ==========================================================================
@@ -287,7 +287,7 @@ SELECT * from testUserData;
4 rows in set
!ok
-SET 'sql-client.display.max-column-width' = '10';
+SET 'table.display.max-column-width' = '10';
[INFO] Execute statement succeeded.
!info
@@ -303,7 +303,7 @@ SELECT * from testUserData;
4 rows in set
!ok
-SET 'sql-client.display.max-column-width' = '40';
+SET 'table.display.max-column-width' = '40';
[INFO] Execute statement succeeded.
!info
@@ -324,7 +324,7 @@ DROP TEMPORARY VIEW testUserData;
[INFO] Execute statement succeeded.
!info
-SET 'sql-client.display.max-column-width' = '30';
+SET 'table.display.max-column-width' = '30';
[INFO] Execute statement succeeded.
!info
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/select_batch.q b/flink-table/flink-sql-client/src/test/resources/sql/select_batch.q
index 7df10767699..1a1e48b3520 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/select_batch.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/select_batch.q
@@ -60,7 +60,7 @@ FROM (VALUES
!ok
# ==========================================================================
-# Testing behavior of sql-client.display.max-column-width
+# Testing behavior of table.display.max-column-width
# Only variable width columns are impacted at the moment => STRING, but not TIMESTAMP nor BOOLEAN
# ==========================================================================
@@ -87,7 +87,7 @@ SELECT * from testUserData;
4 rows in set
!ok
-SET 'sql-client.display.max-column-width' = '10';
+SET 'table.display.max-column-width' = '10';
[INFO] Execute statement succeeded.
!info
@@ -103,7 +103,7 @@ SELECT * from testUserData;
4 rows in set
!ok
-SET 'sql-client.display.max-column-width' = '40';
+SET 'table.display.max-column-width' = '40';
[INFO] Execute statement succeeded.
!info
@@ -124,7 +124,7 @@ DROP TEMPORARY VIEW testUserData;
[INFO] Execute statement succeeded.
!info
-SET 'sql-client.display.max-column-width' = '30';
+SET 'table.display.max-column-width' = '30';
[INFO] Execute statement succeeded.
!info
diff --git a/flink-table/flink-sql-client/src/test/resources/sql/table.q b/flink-table/flink-sql-client/src/test/resources/sql/table.q
index 02168b1c6ac..4a1ed78d4e8 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/table.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/table.q
@@ -148,7 +148,7 @@ show columns from orders;
5 rows in set
!ok
-SET 'sql-client.display.max-column-width' = '10';
+SET 'table.display.max-column-width' = '10';
[INFO] Execute statement succeeded.
!info
@@ -165,7 +165,7 @@ show columns from orders;
5 rows in set
!ok
-SET 'sql-client.display.max-column-width' = '100';
+SET 'table.display.max-column-width' = '100';
[INFO] Execute statement succeeded.
!info
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java
index d5e94678d69..f15f296fc96 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java
@@ -102,28 +102,6 @@ public class OptimizerConfigOptions {
+ TABLE_OPTIMIZER_REUSE_SUB_PLAN_ENABLED.key()
+ " is true.");
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH)
- public static final ConfigOption<Boolean> TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED =
- key("table.optimizer.source.aggregate-pushdown-enabled")
- .booleanType()
- .defaultValue(true)
- .withDescription(
- "When it is true, the optimizer will push down the local aggregates into "
- + "the TableSource which implements SupportsAggregatePushDown.");
-
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
- public static final ConfigOption<Boolean> TABLE_OPTIMIZER_SOURCE_PREDICATE_PUSHDOWN_ENABLED =
- key("table.optimizer.source.predicate-pushdown-enabled")
- .booleanType()
- .defaultValue(true)
- .withDescription(
- "When it is true, the optimizer will push down predicates into the FilterableTableSource. "
- + "Default value is true.");
-
@Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
public static final ConfigOption<Boolean> TABLE_OPTIMIZER_SOURCE_REPORT_STATISTICS_ENABLED =
key("table.optimizer.source.report-statistics-enabled")
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java
deleted file mode 100644
index 5bb191ec056..00000000000
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/LookupJoinHintOptions.java
+++ /dev/null
@@ -1,203 +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.flink.table.planner.hint;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.table.api.config.ExecutionConfigOptions;
-
-import org.apache.flink.shaded.guava32.com.google.common.collect.ImmutableSet;
-
-import java.time.Duration;
-import java.util.HashSet;
-import java.util.Set;
-
-import static org.apache.flink.configuration.ConfigOptions.key;
-
-/**
- * This {@link LookupJoinHintOptions} defines valid hint options of lookup join hint.
- *
- * @deprecated The configurations have been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link org.apache.flink.table.api.config.LookupJoinHintOptions}
- * instead.
- */
-@Deprecated
-@Internal
-public class LookupJoinHintOptions {
- private LookupJoinHintOptions() {}
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#LOOKUP_TABLE} instead.
- */
- @Deprecated
- public static final ConfigOption<String> LOOKUP_TABLE =
- key("table")
- .stringType()
- .noDefaultValue()
- .withDescription("The table name of the lookup source.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#ASYNC_LOOKUP} instead.
- */
- @Deprecated
- public static final ConfigOption<Boolean> ASYNC_LOOKUP =
- key("async")
- .booleanType()
- .noDefaultValue()
- .withDescription(
- "Value can be 'true' or 'false' to suggest the planner choose the corresponding"
- + " lookup function. If the backend lookup source does not support the"
- + " suggested lookup mode, it will take no effect.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#ASYNC_OUTPUT_MODE} instead.
- */
- @Deprecated
- public static final ConfigOption<ExecutionConfigOptions.AsyncOutputMode> ASYNC_OUTPUT_MODE =
- key("output-mode")
- .enumType(ExecutionConfigOptions.AsyncOutputMode.class)
- .noDefaultValue()
- .withDescription(
- "Output mode for asynchronous operations which will convert to {@see AsyncDataStream.OutputMode}, ORDERED by default. "
- + "If set to ALLOW_UNORDERED, will attempt to use {@see AsyncDataStream.OutputMode.UNORDERED} when it does not "
- + "affect the correctness of the result, otherwise ORDERED will be still used.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#ASYNC_CAPACITY} instead.
- */
- @Deprecated
- public static final ConfigOption<Integer> ASYNC_CAPACITY =
- key("capacity")
- .intType()
- .noDefaultValue()
- .withDescription(
- "The max number of async i/o operation that the async lookup join can trigger.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#ASYNC_TIMEOUT} instead.
- */
- @Deprecated
- public static final ConfigOption<Duration> ASYNC_TIMEOUT =
- key("timeout")
- .durationType()
- .noDefaultValue()
- .withDescription(
- "Timeout from first invoke to final completion of asynchronous operation, may include multiple"
- + " retries, and will be reset in case of failover.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#RETRY_PREDICATE} instead.
- */
- @Deprecated
- public static final ConfigOption<String> RETRY_PREDICATE =
- key("retry-predicate")
- .stringType()
- .noDefaultValue()
- .withDescription(
- "A predicate expresses the retry condition, can be 'lookup-miss' which will"
- + " enable retry if lookup result is empty.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#RETRY_STRATEGY} instead.
- */
- @Deprecated
- public static final ConfigOption<RetryStrategy> RETRY_STRATEGY =
- key("retry-strategy")
- .enumType(RetryStrategy.class)
- .noDefaultValue()
- .withDescription("The retry strategy name, can be 'fixed-delay' for now.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#FIXED_DELAY} instead.
- */
- @Deprecated
- public static final ConfigOption<Duration> FIXED_DELAY =
- key("fixed-delay")
- .durationType()
- .noDefaultValue()
- .withDescription("Delay time for the 'fixed-delay' retry strategy.");
-
- /**
- * @deprecated This configuration has been deprecated as part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions#MAX_ATTEMPTS} instead.
- */
- @Deprecated
- public static final ConfigOption<Integer> MAX_ATTEMPTS =
- key("max-attempts")
- .intType()
- .noDefaultValue()
- .withDescription("Max attempt number of the 'fixed-delay' retry strategy.");
-
- public static final String LOOKUP_MISS_PREDICATE = "lookup_miss";
-
- private static Set<ConfigOption> requiredKeys = new HashSet<>();
- private static Set<ConfigOption> supportedKeys = new HashSet<>();
-
- static {
- requiredKeys.add(LOOKUP_TABLE);
-
- supportedKeys.add(LOOKUP_TABLE);
- supportedKeys.add(ASYNC_LOOKUP);
- supportedKeys.add(ASYNC_CAPACITY);
- supportedKeys.add(ASYNC_TIMEOUT);
- supportedKeys.add(ASYNC_OUTPUT_MODE);
- supportedKeys.add(RETRY_PREDICATE);
- supportedKeys.add(RETRY_STRATEGY);
- supportedKeys.add(FIXED_DELAY);
- supportedKeys.add(MAX_ATTEMPTS);
- }
-
- public static ImmutableSet<ConfigOption> getRequiredOptions() {
- return ImmutableSet.copyOf(requiredKeys);
- }
-
- public static ImmutableSet<ConfigOption> getSupportedOptions() {
- return ImmutableSet.copyOf(supportedKeys);
- }
-
- /**
- * Supported retry strategies.
- *
- * @deprecated This strategy has been deprecated as a part of FLIP-457 and will be removed in
- * Flink 2.0. Please use {@link
- * org.apache.flink.table.api.config.LookupJoinHintOptions.RetryStrategy} instead.
- */
- @Deprecated
- public enum RetryStrategy {
- /** Fixed-delay retry strategy. */
- FIXED_DELAY
- }
-}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRule.java
index 2a4045dbfe8..6762d280c70 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoLegacyTableSourceScanRule.java
@@ -18,16 +18,13 @@
package org.apache.flink.table.planner.plan.rules.logical;
-import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableException;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.planner.calcite.FlinkContext;
import org.apache.flink.table.planner.expressions.converter.ExpressionConverter;
import org.apache.flink.table.planner.plan.schema.FlinkPreparingTableBase;
import org.apache.flink.table.planner.plan.schema.LegacyTableSourceTable;
import org.apache.flink.table.planner.plan.stats.FlinkStatistic;
-import org.apache.flink.table.planner.plan.utils.FlinkRelOptUtil;
import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
import org.apache.flink.table.planner.plan.utils.RexNodeExtractor;
import org.apache.flink.table.sources.FilterableTableSource;
@@ -50,7 +47,6 @@ import java.util.stream.Collectors;
import scala.Tuple2;
import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapContext;
-import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
/** Planner rule that tries to push a filter into a {@link FilterableTableSource}. */
@Value.Enclosing
@@ -71,12 +67,6 @@ public class PushFilterIntoLegacyTableSourceScanRule
@Override
public boolean matches(RelOptRuleCall call) {
- TableConfig tableConfig = unwrapTableConfig(call);
- if (!tableConfig.get(
- OptimizerConfigOptions.TABLE_OPTIMIZER_SOURCE_PREDICATE_PUSHDOWN_ENABLED)) {
- return false;
- }
-
Filter filter = call.rel(0);
if (filter.getCondition() == null) {
return false;
@@ -110,11 +100,9 @@ public class PushFilterIntoLegacyTableSourceScanRule
FlinkPreparingTableBase relOptTable) {
RelBuilder relBuilder = call.builder();
FlinkContext context = unwrapContext(call);
- int maxCnfNodeCount = FlinkRelOptUtil.getMaxCnfNodeCount(scan);
Tuple2<Expression[], RexNode[]> extracted =
RexNodeExtractor.extractConjunctiveConditions(
filter.getCondition(),
- maxCnfNodeCount,
filter.getInput().getRowType().getFieldNames(),
relBuilder.getRexBuilder(),
context.getFunctionCatalog(),
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java
index eb5e4f6d25d..76e5a9b4a10 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java
@@ -17,8 +17,6 @@
package org.apache.flink.table.planner.plan.rules.logical;
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
import org.apache.flink.table.expressions.ResolvedExpression;
@@ -27,10 +25,8 @@ import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
import org.apache.flink.table.planner.plan.abilities.source.SourceAbilityContext;
import org.apache.flink.table.planner.plan.abilities.source.SourceAbilitySpec;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;
-import org.apache.flink.table.planner.utils.ShortcutUtils;
import org.apache.calcite.plan.RelOptRule;
-import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptRuleOperand;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rex.RexNode;
@@ -48,13 +44,6 @@ public abstract class PushFilterIntoSourceScanRuleBase extends RelOptRule {
super(operand, description);
}
- @Override
- public boolean matches(RelOptRuleCall call) {
- TableConfig tableConfig = ShortcutUtils.unwrapContext(call.getPlanner()).getTableConfig();
- return tableConfig.get(
- OptimizerConfigOptions.TABLE_OPTIMIZER_SOURCE_PREDICATE_PUSHDOWN_ENABLED);
- }
-
protected List<RexNode> convertExpressionToRexNode(
List<ResolvedExpression> expressions, RelBuilder relBuilder) {
ExpressionConverter exprConverter = new ExpressionConverter(relBuilder);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java
index ea39984c56c..236d04b7f23 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoTableSourceScanRule.java
@@ -38,7 +38,6 @@ import org.apache.flink.table.planner.plan.abilities.source.PartitionPushDownSpe
import org.apache.flink.table.planner.plan.abilities.source.SourceAbilityContext;
import org.apache.flink.table.planner.plan.abilities.source.SourceAbilitySpec;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;
-import org.apache.flink.table.planner.plan.utils.FlinkRelOptUtil;
import org.apache.flink.table.planner.plan.utils.PartitionPruner;
import org.apache.flink.table.planner.plan.utils.RexNodeExtractor;
import org.apache.flink.table.planner.plan.utils.RexNodeToExpressionConverter;
@@ -126,7 +125,6 @@ public class PushPartitionIntoTableSourceScanRule extends RelOptRule {
Tuple2<Seq<RexNode>, Seq<RexNode>> allPredicates =
RexNodeExtractor.extractPartitionPredicateList(
filter.getCondition(),
- FlinkRelOptUtil.getMaxCnfNodeCount(scan),
inputFieldNames.toArray(new String[0]),
rexBuilder,
partitionFieldNames.toArray(new String[0]));
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java
index c270c11b71b..04832a80798 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/SubQueryDecorrelator.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.planner.plan.rules.logical;
import org.apache.flink.table.planner.calcite.FlinkRelBuilder;
-import org.apache.flink.table.planner.plan.utils.FlinkRelOptUtil;
import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
import org.apache.flink.util.Preconditions;
@@ -123,9 +122,7 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
* @return Decorrelate result.
*/
public static Result decorrelateQuery(RelNode rootRel) {
- int maxCnfNodeCount = FlinkRelOptUtil.getMaxCnfNodeCount(rootRel);
-
- final CorelMapBuilder builder = new CorelMapBuilder(maxCnfNodeCount);
+ final CorelMapBuilder builder = new CorelMapBuilder();
final CorelMap corelMap = builder.build(rootRel);
if (builder.hasNestedCorScope || builder.hasUnsupportedCorCondition) {
return null;
@@ -141,9 +138,7 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
final SubQueryDecorrelator decorrelator =
new SubQueryDecorrelator(
- new SubQueryRelDecorrelator(
- corelMap, relBuilder, rexBuilder, maxCnfNodeCount),
- relBuilder);
+ new SubQueryRelDecorrelator(corelMap, relBuilder, rexBuilder), relBuilder);
rootRel.accept(decorrelator);
return new Result(decorrelator.subQueryMap);
@@ -264,13 +259,12 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
final Set<CorrelationId> variableSet,
final RexNode condition,
final RexBuilder rexBuilder,
- final int maxCnfNodeCount,
final List<RexNode> corConditions,
final List<RexNode> nonCorConditions,
final List<RexNode> unsupportedCorConditions) {
// converts the expanded expression to conjunctive normal form,
// like "(a AND b) OR c" will be converted to "(a OR c) AND (b OR c)"
- final RexNode cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, condition);
+ final RexNode cnf = FlinkRexUtil.toCnf(rexBuilder, condition);
// converts the cnf condition to a list of AND conditions
final List<RexNode> conjunctions = RelOptUtil.conjunctions(cnf);
// `true` for RexNode is supported correlation condition,
@@ -418,14 +412,11 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
private final ReflectUtil.MethodDispatcher<Frame> dispatcher =
ReflectUtil.createMethodDispatcher(
Frame.class, this, "decorrelateRel", RelNode.class);
- private final int maxCnfNodeCount;
- SubQueryRelDecorrelator(
- CorelMap cm, RelBuilder relBuilder, RexBuilder rexBuilder, int maxCnfNodeCount) {
+ SubQueryRelDecorrelator(CorelMap cm, RelBuilder relBuilder, RexBuilder rexBuilder) {
this.cm = cm;
this.relBuilder = relBuilder;
this.rexBuilder = rexBuilder;
- this.maxCnfNodeCount = maxCnfNodeCount;
}
Frame getInvoke(RelNode r) {
@@ -536,7 +527,6 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
cm.mapSubQueryNodeToCorSet.get(rel),
rel.getCondition(),
rexBuilder,
- maxCnfNodeCount,
corConditions,
nonCorConditions,
unsupportedCorConditions);
@@ -958,7 +948,7 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
/** Builds a {@link CorelMap}. */
private static class CorelMapBuilder extends RelShuttleImpl {
- private final int maxCnfNodeCount;
+
// nested correlation variables in SubQuery, such as:
// SELECT * FROM t1 WHERE EXISTS (SELECT * FROM t2 WHERE t1.a = t2.c AND
// t2.d IN (SELECT t3.d FROM t3 WHERE t1.b = t3.e)
@@ -977,10 +967,6 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
// true if SubQuery rel tree has Over node, else false.
boolean hasOverNode = false;
- public CorelMapBuilder(int maxCnfNodeCount) {
- this.maxCnfNodeCount = maxCnfNodeCount;
- }
-
final SortedMap<CorrelationId, RelNode> mapCorToCorRel = new TreeMap<>();
final com.google.common.collect.SortedSetMultimap<RelNode, CorRef> mapRefRelToCorRef =
com.google.common.collect.Multimaps.newSortedSetMultimap(
@@ -1154,7 +1140,6 @@ public class SubQueryDecorrelator extends RelShuttleImpl {
mapSubQueryNodeToCorSet.get(filter),
filter.getCondition(),
filter.getCluster().getRexBuilder(),
- maxCnfNodeCount,
corConditions,
new ArrayList<>(),
unsupportedCorConditions);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java
index c576e5b684d..88cbecf618f 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java
@@ -18,8 +18,6 @@
package org.apache.flink.table.planner.plan.rules.physical.batch;
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
@@ -35,7 +33,6 @@ import org.apache.flink.table.planner.plan.schema.TableSourceTable;
import org.apache.flink.table.planner.plan.stats.FlinkStatistic;
import org.apache.flink.table.planner.plan.utils.RexNodeExtractor;
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
-import org.apache.flink.table.planner.utils.ShortcutUtils;
import org.apache.flink.table.types.logical.RowType;
import org.apache.calcite.plan.RelOptRule;
@@ -77,16 +74,9 @@ public abstract class PushLocalAggIntoScanRuleBase extends RelOptRule {
}
protected boolean canPushDown(
- RelOptRuleCall call,
BatchPhysicalGroupAggregateBase aggregate,
BatchPhysicalTableSourceScan tableSourceScan,
@Nullable BatchPhysicalCalc calc) {
- TableConfig tableConfig = ShortcutUtils.unwrapContext(call.getPlanner()).getTableConfig();
- if (!tableConfig.get(
- OptimizerConfigOptions.TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED)) {
- return false;
- }
-
if (aggregate.isFinal() || aggregate.getAggCallList().isEmpty()) {
return false;
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java
index e800a77a56f..4e7cbbc459d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules.physical.batch;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalExchange;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalLocalHashAggregate;
@@ -30,8 +29,7 @@ import org.apache.calcite.plan.RelOptRuleCall;
/**
* Planner rule that tries to push a local hash aggregate which without sort into a {@link
* BatchPhysicalTableSourceScan} whose table is a {@link TableSourceTable} with a source supporting
- * {@link SupportsAggregatePushDown}. The {@link
- * OptimizerConfigOptions#TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED} need to be true.
+ * {@link SupportsAggregatePushDown}.
*
* <p>Suppose we have the original physical plan:
*
@@ -67,7 +65,7 @@ public class PushLocalHashAggIntoScanRule extends PushLocalAggIntoScanRuleBase {
public boolean matches(RelOptRuleCall call) {
BatchPhysicalLocalHashAggregate localAggregate = call.rel(1);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(2);
- return canPushDown(call, localAggregate, tableSourceScan, null);
+ return canPushDown(localAggregate, tableSourceScan, null);
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java
index 2ec53eb5304..12c8e8e8980 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules.physical.batch;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalExchange;
@@ -31,8 +30,7 @@ import org.apache.calcite.plan.RelOptRuleCall;
/**
* Planner rule that tries to push a local hash aggregate which with calc into a {@link
* BatchPhysicalTableSourceScan} whose table is a {@link TableSourceTable} with a source supporting
- * {@link SupportsAggregatePushDown}. The {@link
- * OptimizerConfigOptions#TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED} need to be true.
+ * {@link SupportsAggregatePushDown}.
*
* <p>Suppose we have the original physical plan:
*
@@ -73,7 +71,7 @@ public class PushLocalHashAggWithCalcIntoScanRule extends PushLocalAggIntoScanRu
BatchPhysicalLocalHashAggregate localHashAgg = call.rel(1);
BatchPhysicalCalc calc = call.rel(2);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(3);
- return canPushDown(call, localHashAgg, tableSourceScan, calc);
+ return canPushDown(localHashAgg, tableSourceScan, calc);
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java
index fe2a2738035..f5c6843f23e 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules.physical.batch;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalExchange;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalLocalSortAggregate;
@@ -30,8 +29,7 @@ import org.apache.calcite.plan.RelOptRuleCall;
/**
* Planner rule that tries to push a local sort aggregate which without sort into a {@link
* BatchPhysicalTableSourceScan} whose table is a {@link TableSourceTable} with a source supporting
- * {@link SupportsAggregatePushDown}. The {@link
- * OptimizerConfigOptions#TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED} need to be true.
+ * {@link SupportsAggregatePushDown}.
*
* <p>Suppose we have the original physical plan:
*
@@ -67,7 +65,7 @@ public class PushLocalSortAggIntoScanRule extends PushLocalAggIntoScanRuleBase {
public boolean matches(RelOptRuleCall call) {
BatchPhysicalLocalSortAggregate localAggregate = call.rel(1);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(2);
- return canPushDown(call, localAggregate, tableSourceScan, null);
+ return canPushDown(localAggregate, tableSourceScan, null);
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java
index a61d88db6d9..11906e8656b 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules.physical.batch;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalExchange;
@@ -31,8 +30,7 @@ import org.apache.calcite.plan.RelOptRuleCall;
/**
* Planner rule that tries to push a local sort aggregate which without sort into a {@link
* BatchPhysicalTableSourceScan} whose table is a {@link TableSourceTable} with a source supporting
- * {@link SupportsAggregatePushDown}. The {@link
- * OptimizerConfigOptions#TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED} need to be true.
+ * {@link SupportsAggregatePushDown}.
*
* <p>Suppose we have the original physical plan:
*
@@ -73,7 +71,7 @@ public class PushLocalSortAggWithCalcIntoScanRule extends PushLocalAggIntoScanRu
BatchPhysicalLocalSortAggregate localAggregate = call.rel(1);
BatchPhysicalCalc calc = call.rel(2);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(3);
- return canPushDown(call, localAggregate, tableSourceScan, calc);
+ return canPushDown(localAggregate, tableSourceScan, calc);
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java
index 0bc7fca86aa..eeeb266f483 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules.physical.batch;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalExchange;
@@ -33,8 +32,7 @@ import org.apache.calcite.plan.RelOptRuleCall;
/**
* Planner rule that tries to push a local sort aggregate which with sort and calc into a {@link
* BatchPhysicalTableSourceScan} whose table is a {@link TableSourceTable} with a source supporting
- * {@link SupportsAggregatePushDown}. The {@link
- * OptimizerConfigOptions#TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED} need to be true.
+ * {@link SupportsAggregatePushDown}.
*
* <p>Suppose we have the original physical plan:
*
@@ -82,7 +80,7 @@ public class PushLocalSortAggWithSortAndCalcIntoScanRule extends PushLocalAggInt
BatchPhysicalGroupAggregateBase localAggregate = call.rel(1);
BatchPhysicalCalc calc = call.rel(3);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(4);
- return canPushDown(call, localAggregate, tableSourceScan, calc);
+ return canPushDown(localAggregate, tableSourceScan, calc);
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java
index 87e433ceabd..014c11e8b49 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules.physical.batch;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalExchange;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalGroupAggregateBase;
@@ -32,8 +31,7 @@ import org.apache.calcite.plan.RelOptRuleCall;
/**
* Planner rule that tries to push a local sort aggregate which with sort into a {@link
* BatchPhysicalTableSourceScan} whose table is a {@link TableSourceTable} with a source supporting
- * {@link SupportsAggregatePushDown}. The {@link
- * OptimizerConfigOptions#TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED} need to be true.
+ * {@link SupportsAggregatePushDown}.
*
* <p>Suppose we have the original physical plan:
*
@@ -75,7 +73,7 @@ public class PushLocalSortAggWithSortIntoScanRule extends PushLocalAggIntoScanRu
public boolean matches(RelOptRuleCall call) {
BatchPhysicalGroupAggregateBase localAggregate = call.rel(1);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(3);
- return canPushDown(call, localAggregate, tableSourceScan, null);
+ return canPushDown(localAggregate, tableSourceScan, null);
}
@Override
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala
index 7ebf2df3286..f51fddc005a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala
@@ -17,16 +17,12 @@
*/
package org.apache.flink.table.planner.codegen.agg.batch
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.streaming.api.operators.OneInputStreamOperator
import org.apache.flink.table.api.config.ExecutionConfigOptions
import org.apache.flink.table.data.{GenericRowData, RowData}
import org.apache.flink.table.data.binary.BinaryRowData
import org.apache.flink.table.data.utils.JoinedRowData
import org.apache.flink.table.functions.DeclarativeAggregateFunction
-import org.apache.flink.table.planner.{JBoolean, JDouble, JLong}
import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, CodeGenUtils, OperatorCodeGenerator, ProjectionCodeGenerator}
import org.apache.flink.table.planner.codegen.CodeGenUtils.ROW_DATA
import org.apache.flink.table.planner.plan.utils.AggregateInfoList
@@ -46,85 +42,6 @@ import org.apache.calcite.tools.RelBuilder
*/
object HashAggCodeGenerator {
- /**
- * Whether to enable adaptive local hash aggregation.
- * @deprecated
- * This configuration has been deprecated as part of FLIP-457 and will be removed in Flink 2.0.
- * Please use
- * [[org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_LOCAL_HASH_AGG_ADAPTIVE_ENABLED]]
- * instead.
- */
- @Deprecated
- @Experimental
- val TABLE_EXEC_LOCAL_HASH_AGG_ADAPTIVE_ENABLED: ConfigOption[JBoolean] =
- key("table.exec.local-hash-agg.adaptive.enabled")
- .booleanType()
- .defaultValue(Boolean.box(true))
- .withDescription(
- s"""
- |Whether to enable adaptive local hash aggregation. Adaptive local hash
- |aggregation is an optimization of local hash aggregation, which can adaptively
- |determine whether to continue to do local hash aggregation according to the distinct
- | value rate of sampling data. If distinct value rate bigger than defined threshold
- |(see parameter: table.exec.local-hash-agg.adaptive.distinct-value-rate-threshold),
- |we will stop aggregating and just send the input data to the downstream after a simple
- |projection. Otherwise, we will continue to do aggregation. Adaptive local hash aggregation
- |only works in batch mode. Default value of this parameter is true.
- |""".stripMargin)
-
- /**
- * The value to define how many records as sample data when adaptive local hash aggregation is
- * enabled.
- * @deprecated
- * This configuration has been deprecated as part of FLIP-457 and will be removed in Flink 2.0.
- * Please use
- * [[org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_LOCAL_HASH_AGG_ADAPTIVE_SAMPLING_THRESHOLD]]
- * instead.
- */
- @Deprecated
- @Experimental
- val TABLE_EXEC_LOCAL_HASH_AGG_ADAPTIVE_SAMPLING_THRESHOLD: ConfigOption[JLong] =
- key("table.exec.local-hash-agg.adaptive.sampling-threshold")
- .longType()
- .defaultValue(Long.box(500000L))
- .withDescription(
- s"""
- |If adaptive local hash aggregation is enabled, this value defines how
- |many records will be used as sampled data to calculate distinct value rate
- |(see parameter: table.exec.local-hash-agg.adaptive.distinct-value-rate-threshold)
- |for the local aggregate. The higher the sampling threshold, the more accurate
- |the distinct value rate is. But as the sampling threshold increases, local
- |aggregation is meaningless when the distinct values rate is low.
- |The default value is 500000.
- |""".stripMargin)
-
- /**
- * The value to define the ratio between local aggregation result number and sampling threshold
- * when adaptive local hash aggregation is enabled.
- *
- * @deprecated
- * This configuration has been deprecated as part of FLIP-457 and will be removed in Flink 2.0.
- * Please use
- * [[org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_LOCAL_HASH_AGG_ADAPTIVE_DISTINCT_VALUE_RATE_THRESHOLD]]
- * instead.
- */
- @Deprecated
- @Experimental
- val TABLE_EXEC_LOCAL_HASH_AGG_ADAPTIVE_DISTINCT_VALUE_RATE_THRESHOLD: ConfigOption[JDouble] =
- key("table.exec.local-hash-agg.adaptive.distinct-value-rate-threshold")
- .doubleType()
- .defaultValue(0.5d)
- .withDescription(
- s"""
- |The distinct value rate can be defined as the number of local
- |aggregation result for the sampled data divided by the sampling
- |threshold (see ${TABLE_EXEC_LOCAL_HASH_AGG_ADAPTIVE_SAMPLING_THRESHOLD.key()}).
- |If the computed result is lower than the given configuration value,
- |the remaining input records proceed to do local aggregation, otherwise
- |the remaining input records are subjected to simple projection which
- |calculation cost is less than local aggregation. The default value is 0.5.
- |""".stripMargin)
-
def genWithKeys(
ctx: CodeGeneratorContext,
builder: RelBuilder,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnNullCount.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnNullCount.scala
index 028277a7d7b..0039432d4d7 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnNullCount.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnNullCount.scala
@@ -21,7 +21,6 @@ import org.apache.flink.table.planner.{JDouble, JList}
import org.apache.flink.table.planner.plan.metadata.FlinkMetadata.ColumnNullCount
import org.apache.flink.table.planner.plan.schema.FlinkPreparingTableBase
import org.apache.flink.table.planner.plan.utils.FlinkRexUtil
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.flink.util.Preconditions
import org.apache.calcite.plan.RelOptUtil
@@ -176,9 +175,7 @@ class FlinkRelMdColumnNullCount private extends MetadataHandler[ColumnNullCount]
} else {
// If predicate has $index is not null, null count of index is must be 0 after predicate.
val rexBuilder = rel.getCluster.getRexBuilder
- val tableConfig = unwrapTableConfig(rel)
- val maxCnfNodeCount = tableConfig.get(FlinkRexUtil.TABLE_OPTIMIZER_CNF_NODES_LIMIT)
- val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, predicate)
+ val cnf = FlinkRexUtil.toCnf(rexBuilder, predicate)
val conjunctions = RelOptUtil.conjunctions(cnf)
val notNullPredicatesAtIndexField = conjunctions.exists {
case call: RexCall if call.getOperator == SqlStdOperatorTable.IS_NOT_NULL =>
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistinctRowCount.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistinctRowCount.scala
index 86158e0c85a..415afa252fc 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistinctRowCount.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistinctRowCount.scala
@@ -23,7 +23,6 @@ import org.apache.flink.table.planner.plan.nodes.calcite.{Expand, Rank, WindowAg
import org.apache.flink.table.planner.plan.nodes.physical.batch._
import org.apache.flink.table.planner.plan.schema.FlinkPreparingTableBase
import org.apache.flink.table.planner.plan.utils.{FlinkRelMdUtil, FlinkRexUtil, RankUtil}
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.plan.volcano.RelSubset
@@ -204,9 +203,7 @@ class FlinkRelMdDistinctRowCount private extends MetadataHandler[BuiltInMetadata
null
} else {
val rexBuilder = rel.getCluster.getRexBuilder
- val tableConfig = unwrapTableConfig(rel)
- val maxCnfNodeCount = tableConfig.get(FlinkRexUtil.TABLE_OPTIMIZER_CNF_NODES_LIMIT)
- val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, predicate)
+ val cnf = FlinkRexUtil.toCnf(rexBuilder, predicate)
val conjunctions = RelOptUtil.conjunctions(cnf)
val conjunctionsWithoutExpandId = conjunctions.filterNot {
c =>
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistribution.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistribution.scala
index 696de982bc4..cc9253829c3 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistribution.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdDistribution.scala
@@ -20,8 +20,6 @@ package org.apache.flink.table.planner.plan.metadata
import org.apache.flink.table.planner.JHashMap
import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution, FlinkRelDistributionTraitDef}
import org.apache.flink.table.planner.plan.metadata.FlinkMetadata.FlinkDistribution
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortRule
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.calcite.rel._
import org.apache.calcite.rel.core.{Calc, Exchange, Sort, TableScan}
@@ -73,18 +71,7 @@ class FlinkRelMdDistribution private extends MetadataHandler[FlinkDistribution]
}
def flinkDistribution(sort: Sort, mq: RelMetadataQuery): FlinkRelDistribution = {
- val tableConfig = unwrapTableConfig(sort)
- val enableRangeSort = tableConfig.get(BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED)
- if (
- (sort.getCollation.getFieldCollations.nonEmpty &&
- sort.fetch == null && sort.offset == null) && enableRangeSort
- ) {
- // If Sort is global sort, and the table config allows the range partition.
- // Then the Sort's required traits will are range distribution and sort collation.
- FlinkRelDistribution.range(sort.getCollation.getFieldCollations)
- } else {
- FlinkRelDistribution.SINGLETON
- }
+ FlinkRelDistribution.SINGLETON
}
def flinkDistribution(exchange: Exchange, mq: RelMetadataQuery): FlinkRelDistribution = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala
index d821feef585..c1f80747dc2 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala
@@ -17,16 +17,13 @@
*/
package org.apache.flink.table.planner.plan.metadata
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.table.planner.plan.logical.{LogicalWindow, SlidingGroupWindow, TumblingGroupWindow}
+import org.apache.flink.table.planner.plan.metadata.FlinkRelMdRowCount.ROWS_PER_LOCAL_AGG
import org.apache.flink.table.planner.plan.nodes.calcite.{Expand, Rank, WindowAggregate}
import org.apache.flink.table.planner.plan.nodes.physical.batch._
import org.apache.flink.table.planner.plan.stats.ValueInterval
import org.apache.flink.table.planner.plan.utils.{FlinkRelMdUtil, SortUtil}
import org.apache.flink.table.planner.plan.utils.AggregateUtil.{hasTimeIntervalType, toLong}
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.calcite.adapter.enumerable.EnumerableLimit
import org.apache.calcite.plan.volcano.RelSubset
@@ -36,7 +33,7 @@ import org.apache.calcite.rel.metadata._
import org.apache.calcite.rex.{RexLiteral, RexNode}
import org.apache.calcite.util._
-import java.lang.{Double => JDouble, Long => JLong}
+import java.lang.{Double => JDouble}
import scala.collection.JavaConversions._
@@ -178,9 +175,7 @@ class FlinkRelMdRowCount private extends MetadataHandler[BuiltInMetadata.RowCoun
ndvOfGroupKeysOnGlobalAgg
} else {
val inputRowCnt = mq.getRowCount(input)
- val tableConfig = unwrapTableConfig(rel)
- val parallelism = (inputRowCnt /
- tableConfig.get(FlinkRelMdRowCount.TABLE_OPTIMIZER_ROWS_PER_LOCALAGG) + 1).toInt
+ val parallelism = (inputRowCnt / ROWS_PER_LOCAL_AGG + 1).toInt
if (parallelism == 1) {
ndvOfGroupKeysOnGlobalAgg
} else if (grouping.isEmpty) {
@@ -451,14 +446,8 @@ object FlinkRelMdRowCount {
val SOURCE: RelMetadataProvider =
ReflectiveRelMetadataProvider.reflectiveSource(BuiltInMethod.ROW_COUNT.method, INSTANCE)
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_ROWS_PER_LOCALAGG: ConfigOption[JLong] =
- key("table.optimizer.rows-per-local-agg")
- .longType()
- .defaultValue(JLong.valueOf(1000000L))
- .withDescription("Sets estimated number of records that one local-agg processes. " +
- "Optimizer will infer whether to use local/global aggregate according to it.")
+ // the estimated number of records that one local-agg processes
+ // optimizer will infer whether to use local/global aggregate according to it
+ val ROWS_PER_LOCAL_AGG = 1000000L
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimator.scala
index aa690fc244f..8aae5799253 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimator.scala
@@ -58,7 +58,6 @@ class SelectivityEstimator(rel: RelNode, mq: FlinkRelMetadataQuery)
private val rexBuilder = rel.getCluster.getRexBuilder
private val tableConfig = unwrapTableConfig(rel)
- private val maxCnfNodeCount = tableConfig.get(FlinkRexUtil.TABLE_OPTIMIZER_CNF_NODES_LIMIT)
// these default values is referred to RelMdUtil#guessSelectivity
private[flink] val defaultComparisonSelectivity = Some(0.5d)
@@ -225,7 +224,7 @@ class SelectivityEstimator(rel: RelNode, mq: FlinkRelMetadataQuery)
* 40.0` and `CAST(b AS INTEGER) > 30`
*/
private def splitAndPredicate(call: RexCall): Seq[RexNode] = {
- val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, call)
+ val cnf = FlinkRexUtil.toCnf(rexBuilder, call)
val conjunctions = RelOptUtil.conjunctions(cnf)
val combinations = new JHashMap[String, JArrayList[RexNode]]()
@@ -330,7 +329,7 @@ class SelectivityEstimator(rel: RelNode, mq: FlinkRelMetadataQuery)
* None if the condition is not supported.
*/
private def estimateAndPredicate(andPredicate: RexCall): Option[Double] = {
- val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, andPredicate)
+ val cnf = FlinkRexUtil.toCnf(rexBuilder, andPredicate)
val conjunctions = RelOptUtil.conjunctions(cnf)
val inputRefs = new JHashSet[Int]()
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalSort.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalSort.scala
index f0b8ed02bc5..cbe6ad93898 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalSort.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalSort.scala
@@ -19,7 +19,6 @@ package org.apache.flink.table.planner.plan.nodes.logical
import org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_SORT_DEFAULT_LIMIT
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED
import org.apache.flink.table.planner.plan.utils.SortUtil
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
@@ -96,13 +95,9 @@ class FlinkLogicalSortBatchConverter(config: Config) extends ConverterRule(confi
val sort = rel.asInstanceOf[LogicalSort]
val newInput = RelOptRule.convert(sort.getInput, FlinkConventions.LOGICAL)
val tableConfig = unwrapTableConfig(sort)
- val enableRangeSort = tableConfig.get(TABLE_EXEC_RANGE_SORT_ENABLED)
val limitValue = tableConfig.get(TABLE_EXEC_SORT_DEFAULT_LIMIT)
val (offset, fetch) =
- if (
- sort.fetch == null && sort.offset == null
- && !enableRangeSort && limitValue > 0
- ) {
+ if (sort.fetch == null && sort.offset == null && limitValue > 0) {
// force the sort add limit
val rexBuilder = rel.getCluster.getRexBuilder
val intType = rexBuilder.getTypeFactory.createSqlType(SqlTypeName.INTEGER)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala
index e22cf17d735..466a019fc4c 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala
@@ -22,7 +22,6 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution, FlinkRelDistributionTraitDef}
import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecHashAggregate
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase
import org.apache.flink.table.planner.plan.utils.RelExplainUtil
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
@@ -114,11 +113,7 @@ class BatchPhysicalHashAggregate(
// so return true if shuffleKeys(Hash[a, b]) start with groupKeys(Hash[a])
true
} else {
- // If partialKey is enabled, try to use partial key to satisfy the required distribution
- val tableConfig = unwrapTableConfig(this)
- val partialKeyEnabled = tableConfig.get(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED)
- partialKeyEnabled && groupKeysList.containsAll(shuffleKeys)
+ false
}
case _ => false
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalOverAggregateBase.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalOverAggregateBase.scala
index 230a05f32c4..ce19f918dbd 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalOverAggregateBase.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalOverAggregateBase.scala
@@ -21,10 +21,8 @@ import org.apache.flink.table.api.TableException
import org.apache.flink.table.planner.CalcitePair
import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution, FlinkRelDistributionTraitDef}
import org.apache.flink.table.planner.plan.cost.{FlinkCost, FlinkCostFactory}
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase
import org.apache.flink.table.planner.plan.utils.{OverAggregateUtil, RelExplainUtil}
import org.apache.flink.table.planner.plan.utils.OverAggregateUtil.splitOutOffsetOrInsensitiveGroup
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.calcite.plan._
import org.apache.calcite.rel._
@@ -140,15 +138,7 @@ abstract class BatchPhysicalOverAggregateBase(
} else {
val isAllFieldsFromInput = requiredDistribution.getKeys.forall(_ < inputFieldCnt)
if (isAllFieldsFromInput) {
- val tableConfig = unwrapTableConfig(this)
- if (
- tableConfig.get(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED)
- ) {
- ImmutableIntList.of(partitionKeyIndices: _*).containsAll(requiredDistribution.getKeys)
- } else {
- requiredDistribution.getKeys == ImmutableIntList.of(partitionKeyIndices: _*)
- }
+ requiredDistribution.getKeys == ImmutableIntList.of(partitionKeyIndices: _*)
} else {
// If requirement distribution keys are not all comes from input directly,
// cannot satisfy requirement distribution and collations.
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalPythonGroupAggregate.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalPythonGroupAggregate.scala
index 1a3b31e13dd..6a269dcfe96 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalPythonGroupAggregate.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalPythonGroupAggregate.scala
@@ -22,7 +22,6 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution, FlinkRelDistributionTraitDef}
import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecPythonGroupAggregate
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase
import org.apache.flink.table.planner.plan.utils.{FlinkRelOptUtil, RelExplainUtil}
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
@@ -94,11 +93,7 @@ class BatchPhysicalPythonGroupAggregate(
// so return true if shuffleKeys(Hash[a, b]) start with groupKeys(Hash[a])
true
} else {
- // If partialKey is enabled, try to use partial key to satisfy the required distribution
- val tableConfig = unwrapTableConfig(this)
- val partialKeyEnabled = tableConfig.get(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED)
- partialKeyEnabled && groupKeysList.containsAll(shuffleKeys)
+ false
}
case _ => false
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalRank.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalRank.scala
index 38b01231d86..c4442405f4f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalRank.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalRank.scala
@@ -24,7 +24,6 @@ import org.apache.flink.table.planner.plan.cost.{FlinkCost, FlinkCostFactory}
import org.apache.flink.table.planner.plan.nodes.calcite.Rank
import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecRank
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase
import org.apache.flink.table.planner.plan.utils.{FlinkRelOptUtil, RelExplainUtil}
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.flink.table.runtime.operators.rank.{ConstantRankRange, RankRange, RankType}
@@ -133,11 +132,7 @@ class BatchPhysicalRank(
// so return true if shuffleKeys(Hash[a, b]) start with partitionKeyList(Hash[a])
true
} else {
- // If partialKey is enabled, try to use partial key to satisfy the required distribution
- val tableConfig = unwrapTableConfig(this)
- val partialKeyEnabled = tableConfig.get(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED)
- partialKeyEnabled && partitionKeyList.containsAll(shuffleKeys)
+ false
}
case _ => false
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortAggregate.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortAggregate.scala
index fb9112f01fc..2dc9c5f3412 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortAggregate.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortAggregate.scala
@@ -22,7 +22,6 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution, FlinkRelDistributionTraitDef}
import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecSortAggregate
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase
import org.apache.flink.table.planner.plan.utils.{FlinkRelOptUtil, RelExplainUtil}
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
@@ -114,11 +113,7 @@ class BatchPhysicalSortAggregate(
// so return true if shuffleKeys(Hash[a, b]) start with groupKeys(Hash[a])
true
} else {
- // If partialKey is enabled, try to use partial key to satisfy the required distribution
- val tableConfig = unwrapTableConfig(this)
- val partialKeyEnabled = tableConfig.get(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED)
- partialKeyEnabled && groupKeysList.containsAll(shuffleKeys)
+ false
}
case _ => false
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/RelNodeBlock.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/RelNodeBlock.scala
index 0599a1fde80..0c4c18723b4 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/RelNodeBlock.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/RelNodeBlock.scala
@@ -17,9 +17,7 @@
*/
package org.apache.flink.table.planner.plan.optimize
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.{ConfigOption, ReadableConfig}
-import org.apache.flink.configuration.ConfigOptions.key
+import org.apache.flink.configuration.ReadableConfig
import org.apache.flink.table.api.config.OptimizerConfigOptions
import org.apache.flink.table.planner.plan.`trait`.MiniBatchInterval
import org.apache.flink.table.planner.plan.nodes.calcite.LegacySink
@@ -30,10 +28,9 @@ import org.apache.flink.util.Preconditions
import com.google.common.collect.Sets
import org.apache.calcite.rel._
-import org.apache.calcite.rel.core.{Aggregate, Project, Snapshot, TableFunctionScan, Union}
+import org.apache.calcite.rel.core._
import org.apache.calcite.rex.RexNode
-import java.lang.{Boolean => JBoolean}
import java.util
import scala.collection.JavaConversions._
@@ -353,44 +350,6 @@ class RelNodeBlockPlanBuilder private (tableConfig: ReadableConfig) {
object RelNodeBlockPlanBuilder {
- /**
- * Whether to treat union-all node as a breakpoint.
- * @deprecated
- * This configuration has been deprecated as part of FLIP-457 and will be removed in Flink 2.0.
- * Please use
- * [[org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_UNIONALL_AS_BREAKPOINT_ENABLED]]
- * instead.
- */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_UNIONALL_AS_BREAKPOINT_ENABLED: ConfigOption[JBoolean] =
- key("table.optimizer.union-all-as-breakpoint-enabled")
- .booleanType()
- .defaultValue(JBoolean.valueOf(true))
- .withDescription(
- "When true, the optimizer will breakup the graph at union-all node " +
- "when it's a breakpoint. When false, the optimizer will skip the union-all node " +
- "even it's a breakpoint, and will try find the breakpoint in its inputs.")
-
- /**
- * Whether to reuse optimize block based on digest.
- * @deprecated
- * This configuration has been deprecated as part of FLIP-457 and will be removed in Flink 2.0.
- * Please use
- * [[org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_REUSE_OPTIMIZE_BLOCK_WITH_DIGEST_ENABLED]]
- * instead.
- */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_REUSE_OPTIMIZE_BLOCK_WITH_DIGEST_ENABLED: ConfigOption[JBoolean] =
- key("table.optimizer.reuse-optimize-block-with-digest-enabled")
- .booleanType()
- .defaultValue(JBoolean.valueOf(false))
- .withDescription(
- "When true, the optimizer will try to find out duplicated sub-plan by " +
- "digest to build optimize block(a.k.a. common sub-graph). " +
- "Each optimize block will be optimized independently.")
-
/**
* Decompose the [[RelNode]] trees into [[RelNodeBlock]] trees. First, convert LogicalNode trees
* to RelNode trees. Second, reuse same sub-plan in different trees. Third, decompose the RelNode
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRule.scala
index e1c3cdfecf9..f107af3aa40 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRule.scala
@@ -17,11 +17,8 @@
*/
package org.apache.flink.table.planner.plan.rules.logical
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
+import org.apache.flink.table.planner.plan.rules.logical.JoinDeriveNullFilterRule.JOIN_NULL_FILTER_THRESHOLD
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
import org.apache.calcite.plan.RelOptRule.{any, operand}
@@ -32,8 +29,6 @@ import org.apache.calcite.rex.RexNode
import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.util.ImmutableIntList
-import java.lang.{Long => JLong}
-
import scala.collection.JavaConversions._
import scala.collection.mutable
@@ -58,9 +53,6 @@ class JoinDeriveNullFilterRule
val rexBuilder = join.getCluster.getRexBuilder
val mq = FlinkRelMetadataQuery.reuseOrCreate(join.getCluster.getMetadataQuery)
- val tableConfig = unwrapTableConfig(join)
- val minNullCount =
- tableConfig.get(JoinDeriveNullFilterRule.TABLE_OPTIMIZER_JOIN_NULL_FILTER_THRESHOLD)
def createIsNotNullFilter(input: RelNode, keys: ImmutableIntList): RelNode = {
val relBuilder = call.builder()
@@ -68,7 +60,7 @@ class JoinDeriveNullFilterRule
keys.foreach {
key =>
val nullCount = mq.getColumnNullCount(input, key)
- if (nullCount != null && nullCount > minNullCount) {
+ if (nullCount != null && nullCount > JOIN_NULL_FILTER_THRESHOLD) {
filters += relBuilder.call(
SqlStdOperatorTable.IS_NOT_NULL,
rexBuilder.makeInputRef(input, key))
@@ -95,15 +87,8 @@ class JoinDeriveNullFilterRule
object JoinDeriveNullFilterRule {
val INSTANCE = new JoinDeriveNullFilterRule
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_JOIN_NULL_FILTER_THRESHOLD: ConfigOption[JLong] =
- key("table.optimizer.join.null-filter-threshold")
- .longType()
- .defaultValue(JLong.valueOf(2000000L))
- .withDescription(
- "To avoid the impact of null values on the single join node, " +
- "We will add a null filter (possibly be pushed down) before the join to filter" +
- " null values when the source of InnerJoin has nullCount more than this value.")
+ // To avoid the impact of null values on the single join node,
+ // We will add a null filter (possibly be pushed down) before the join to filter
+ // null values when the source of InnerJoin has nullCount more than this value.
+ val JOIN_NULL_FILTER_THRESHOLD = 2000000L
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRule.scala
index 1ae85c17a30..5510c7e85c4 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRule.scala
@@ -25,7 +25,7 @@ import org.apache.flink.table.plan.stats.TableStats
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.schema.LegacyTableSourceTable
import org.apache.flink.table.planner.plan.stats.FlinkStatistic
-import org.apache.flink.table.planner.plan.utils.{FlinkRelOptUtil, PartitionPruner, RexNodeExtractor, RexNodeToExpressionConverter}
+import org.apache.flink.table.planner.plan.utils.{PartitionPruner, RexNodeExtractor, RexNodeToExpressionConverter}
import org.apache.flink.table.planner.utils.CatalogTableStatisticsConverter
import org.apache.flink.table.planner.utils.JavaScalaConversionUtil.toScala
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapContext
@@ -84,11 +84,9 @@ class PushPartitionIntoLegacyTableSourceScanRule
val relBuilder = call.builder()
val rexBuilder = relBuilder.getRexBuilder
- val maxCnfNodeCount = FlinkRelOptUtil.getMaxCnfNodeCount(scan)
val (partitionPredicates, nonPartitionPredicates) =
RexNodeExtractor.extractPartitionPredicateList(
filter.getCondition,
- maxCnfNodeCount,
inputFields,
rexBuilder,
partitionFieldNames
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashJoinRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashJoinRule.scala
index 4e22525e449..71e1ce252c6 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashJoinRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalHashJoinRule.scala
@@ -29,12 +29,9 @@ import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
import org.apache.calcite.plan.RelOptRule.{any, operand}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.core.{Join, JoinRelType}
-import org.apache.calcite.util.ImmutableIntList
import java.util
-import scala.collection.JavaConversions._
-
/**
* Rule that converts [[FlinkLogicalJoin]] to [[BatchPhysicalHashJoin]] if there exists at least one
* equal-join condition and ShuffleHashJoin or BroadcastHashJoin are enabled.
@@ -146,18 +143,6 @@ class BatchPhysicalHashJoinRule
transformToEquiv(
toHashTraitByColumns(joinInfo.leftKeys),
toHashTraitByColumns(joinInfo.rightKeys))
-
- // add more possibility to only shuffle by partial joinKeys, now only single one
- val isShuffleByPartialKeyEnabled =
- tableConfig.get(BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED)
- if (isShuffleByPartialKeyEnabled && joinInfo.pairs().length > 1) {
- joinInfo.pairs().foreach {
- pair =>
- transformToEquiv(
- toHashTraitByColumns(ImmutableIntList.of(pair.source)),
- toHashTraitByColumns(ImmutableIntList.of(pair.target)))
- }
- }
}
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala
index d69925b4ca2..b895eb8cb0f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala
@@ -17,17 +17,13 @@
*/
package org.apache.flink.table.planner.plan.rules.physical.batch
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.table.api.{TableConfig, TableException, ValidationException}
import org.apache.flink.table.api.config.OptimizerConfigOptions
-import org.apache.flink.table.planner.JDouble
import org.apache.flink.table.planner.hint.{FlinkHints, JoinStrategy}
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalLocalHashAggregate
+import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase.SEMI_JOIN_BUILD_DISTINCT_NDV_RATIO
import org.apache.flink.table.planner.plan.utils.{JoinUtil, OperatorType}
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.flink.table.planner.utils.TableConfigUtils.isOperatorDisabled
import org.apache.calcite.plan.RelOptRule
@@ -35,8 +31,6 @@ import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.core.{Join, JoinRelType}
import org.apache.calcite.util.ImmutableBitSet
-import java.lang.{Boolean => JBoolean, Double => JDouble}
-
import scala.collection.JavaConversions._
trait BatchPhysicalJoinRuleBase {
@@ -75,16 +69,13 @@ trait BatchPhysicalJoinRuleBase {
}
def chooseSemiBuildDistinct(buildRel: RelNode, distinctKeys: Seq[Int]): Boolean = {
- val tableConfig = unwrapTableConfig(buildRel)
val mq = buildRel.getCluster.getMetadataQuery
- val ratioConf =
- tableConfig.get(BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SEMI_JOIN_BUILD_DISTINCT_NDV_RATIO)
val inputRows = mq.getRowCount(buildRel)
val ndvOfGroupKey = mq.getDistinctRowCount(buildRel, ImmutableBitSet.of(distinctKeys: _*), null)
if (ndvOfGroupKey == null) {
false
} else {
- ndvOfGroupKey / inputRows < ratioConf
+ ndvOfGroupKey / inputRows < SEMI_JOIN_BUILD_DISTINCT_NDV_RATIO
}
}
@@ -322,31 +313,8 @@ trait BatchPhysicalJoinRuleBase {
}
object BatchPhysicalJoinRuleBase {
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_SEMI_JOIN_BUILD_DISTINCT_NDV_RATIO: ConfigOption[JDouble] =
- key("table.optimizer.semi-anti-join.build-distinct.ndv-ratio")
- .doubleType()
- .defaultValue(JDouble.valueOf(0.8))
- .withDescription(
- "In order to reduce the amount of data on semi/anti join's" +
- " build side, we will add distinct node before semi/anti join when" +
- " the semi-side or semi/anti join can distinct a lot of data in advance." +
- " We add this configuration to help the optimizer to decide whether to" +
- " add the distinct.")
-
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED: ConfigOption[JBoolean] =
- key("table.optimizer.shuffle-by-partial-key-enabled")
- .booleanType()
- .defaultValue(JBoolean.valueOf(false))
- .withDescription(
- "Enables shuffling by partial partition keys. " +
- "For example, A join with join condition: L.c1 = R.c1 and L.c2 = R.c2. " +
- "If this flag is enabled, there are 3 shuffle strategy:\n " +
- "1. L and R shuffle by c1 \n 2. L and R shuffle by c2\n " +
- "3. L and R shuffle by c1 and c2\n It can reduce some shuffle cost someTimes.")
+ // In order to reduce the amount of data on semi/anti join's
+ // build side, we will add distinct node before semi/anti join when
+ // the semi-side or semi/anti join can distinct a lot of data in advance.
+ val SEMI_JOIN_BUILD_DISTINCT_NDV_RATIO = 0.8
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.scala
index f7fae1bab24..61a1833523f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.scala
@@ -17,27 +17,19 @@
*/
package org.apache.flink.table.planner.plan.rules.physical.batch
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.table.planner.hint.JoinStrategy
import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalJoin
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalSortMergeJoin
-import org.apache.flink.table.planner.plan.utils.FlinkRelOptUtil
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
import org.apache.calcite.plan.RelOptRule.{any, operand}
-import org.apache.calcite.rel.{RelCollations, RelNode}
+import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.core.Join
import org.apache.calcite.util.ImmutableIntList
-import java.lang.{Boolean => JBoolean}
-
-import scala.collection.JavaConversions._
-
/**
* Rule that converts [[FlinkLogicalJoin]] to [[BatchPhysicalSortMergeJoin]] if there exists at
* least one equal-join condition and SortMergeJoin is enabled.
@@ -62,97 +54,38 @@ class BatchPhysicalSortMergeJoinRule
def getTraitSetByShuffleKeys(
shuffleKeys: ImmutableIntList,
- requireStrict: Boolean,
- requireCollation: Boolean): RelTraitSet = {
- var traitSet = call.getPlanner
+ requireStrict: Boolean): RelTraitSet = {
+ call.getPlanner
.emptyTraitSet()
.replace(FlinkConventions.BATCH_PHYSICAL)
.replace(FlinkRelDistribution.hash(shuffleKeys, requireStrict))
- if (requireCollation) {
- val fieldCollations = shuffleKeys.map(FlinkRelOptUtil.ofRelFieldCollation(_))
- val relCollation = RelCollations.of(fieldCollations)
- traitSet = traitSet.replace(relCollation)
- }
- traitSet
- }
-
- def transformToEquiv(
- leftRequiredShuffleKeys: ImmutableIntList,
- rightRequiredShuffleKeys: ImmutableIntList,
- requireLeftSorted: Boolean,
- requireRightSorted: Boolean): Unit = {
-
- val leftRequiredTrait =
- getTraitSetByShuffleKeys(leftRequiredShuffleKeys, requireStrict = true, requireLeftSorted)
- val rightRequiredTrait =
- getTraitSetByShuffleKeys(rightRequiredShuffleKeys, requireStrict = true, requireRightSorted)
-
- val newLeft = RelOptRule.convert(left, leftRequiredTrait)
- val newRight = RelOptRule.convert(right, rightRequiredTrait)
-
- val providedTraitSet = call.getPlanner
- .emptyTraitSet()
- .replace(FlinkConventions.BATCH_PHYSICAL)
- val newJoin = new BatchPhysicalSortMergeJoin(
- join.getCluster,
- providedTraitSet,
- newLeft,
- newRight,
- join.getCondition,
- join.getJoinType,
- requireLeftSorted,
- requireRightSorted)
- call.transformTo(newJoin)
}
- val tableConfig = unwrapTableConfig(join)
- val candidates =
- if (tableConfig.get(BatchPhysicalSortMergeJoinRule.TABLE_OPTIMIZER_SMJ_REMOVE_SORT_ENABLED)) {
- // add more possibility to remove redundant sort, and longer optimization time
- Array((false, false), (true, false), (false, true), (true, true))
- } else {
- // will not try to remove redundant sort, and shorter optimization time
- Array((false, false))
- }
- candidates.foreach {
- case (requireLeftSorted, requireRightSorted) =>
- transformToEquiv(
- joinInfo.leftKeys,
- joinInfo.rightKeys,
- requireLeftSorted,
- requireRightSorted)
- }
-
- // add more possibility to only shuffle by partial joinKeys, now only single one
- val isShuffleByPartialKeyEnabled =
- tableConfig.get(BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED)
- if (isShuffleByPartialKeyEnabled && joinInfo.pairs().length > 1) {
- joinInfo.pairs().foreach {
- pair =>
- // sort require full key not partial key,
- // so requireLeftSorted and requireRightSorted should both be false here
- transformToEquiv(
- ImmutableIntList.of(pair.source),
- ImmutableIntList.of(pair.target),
- requireLeftSorted = false,
- requireRightSorted = false)
- }
- }
+ val leftRequiredTrait =
+ getTraitSetByShuffleKeys(joinInfo.leftKeys, requireStrict = true)
+ val rightRequiredTrait =
+ getTraitSetByShuffleKeys(joinInfo.rightKeys, requireStrict = true)
+
+ val newLeft = RelOptRule.convert(left, leftRequiredTrait)
+ val newRight = RelOptRule.convert(right, rightRequiredTrait)
+
+ val providedTraitSet = call.getPlanner
+ .emptyTraitSet()
+ .replace(FlinkConventions.BATCH_PHYSICAL)
+ // do not try to remove redundant sort for shorter optimization time
+ val newJoin = new BatchPhysicalSortMergeJoin(
+ join.getCluster,
+ providedTraitSet,
+ newLeft,
+ newRight,
+ join.getCondition,
+ join.getJoinType,
+ false,
+ false)
+ call.transformTo(newJoin)
}
}
object BatchPhysicalSortMergeJoinRule {
val INSTANCE: RelOptRule = new BatchPhysicalSortMergeJoinRule
-
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_SMJ_REMOVE_SORT_ENABLED: ConfigOption[JBoolean] =
- key("table.optimizer.smj.remove-sort-enabled")
- .booleanType()
- .defaultValue(JBoolean.FALSE)
- .withDescription(
- "When true, the optimizer will try to remove redundant sort " +
- "for sort merge join. However that will increase optimization time. " +
- "Default value is false.")
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortRule.scala
index f02d9fc46b1..621803fb7e4 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortRule.scala
@@ -17,22 +17,16 @@
*/
package org.apache.flink.table.planner.plan.rules.physical.batch
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalSort
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalSort
-import org.apache.flink.table.planner.utils.ShortcutUtils
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.convert.ConverterRule
import org.apache.calcite.rel.convert.ConverterRule.Config
-import java.lang.{Boolean => JBoolean}
-
/**
* Rule that matches [[FlinkLogicalSort]] which sort fields is non-empty and both `fetch` and
* `offset` are null, and converts it to [[BatchPhysicalSort]].
@@ -48,13 +42,7 @@ class BatchPhysicalSortRule(config: Config) extends ConverterRule(config) {
override def convert(rel: RelNode): RelNode = {
val sort: FlinkLogicalSort = rel.asInstanceOf[FlinkLogicalSort]
val input = sort.getInput
- val tableConfig = ShortcutUtils.unwrapTableConfig(sort)
- val enableRangeSort = tableConfig.get(BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED)
- val distribution = if (enableRangeSort) {
- FlinkRelDistribution.range(sort.getCollation.getFieldCollations)
- } else {
- FlinkRelDistribution.SINGLETON
- }
+ val distribution = FlinkRelDistribution.SINGLETON
val requiredTraitSet = input.getTraitSet
.replace(distribution)
.replace(FlinkConventions.BATCH_PHYSICAL)
@@ -74,15 +62,4 @@ object BatchPhysicalSortRule {
FlinkConventions.LOGICAL,
FlinkConventions.BATCH_PHYSICAL,
"BatchPhysicalSortRule"))
-
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Experimental
- val TABLE_EXEC_RANGE_SORT_ENABLED: ConfigOption[JBoolean] =
- key("table.exec.range-sort.enabled")
- .booleanType()
- .defaultValue(JBoolean.valueOf(false))
- .withDescription("Sets whether to enable range sort, use range sort to sort all data in" +
- " several partitions. When it is false, sorting in only one partition")
-
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/IncrementalAggregateRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/IncrementalAggregateRule.scala
index c40f7bc83e0..0b788427b65 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/IncrementalAggregateRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/IncrementalAggregateRule.scala
@@ -17,9 +17,6 @@
*/
package org.apache.flink.table.planner.plan.rules.physical.stream
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.table.api.config.OptimizerConfigOptions
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.PartialFinalType
@@ -31,7 +28,6 @@ import org.apache.flink.util.Preconditions
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelOptUtil}
import org.apache.calcite.plan.RelOptRule.{any, operand}
-import java.lang.{Boolean => JBoolean}
import java.util.Collections
/**
@@ -155,27 +151,4 @@ class IncrementalAggregateRule
object IncrementalAggregateRule {
val INSTANCE = new IncrementalAggregateRule
-
- /**
- * Whether to enable incremental aggregation.
- *
- * @deprecated
- * This configuration has been deprecated as part of FLIP-457 and will be removed in Flink 2.0.
- * Please use
- * [[org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_INCREMENTAL_AGG_ENABLED]]
- * instead.
- */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_INCREMENTAL_AGG_ENABLED: ConfigOption[JBoolean] =
- key("table.optimizer.incremental-agg-enabled")
- .booleanType()
- .defaultValue(JBoolean.valueOf(true))
- .withDescription(
- "When both local aggregation and distinct aggregation splitting " +
- "are enabled, a distinct aggregation will be optimized into four aggregations, " +
- "i.e., local-agg1, global-agg1, local-agg2 and global-Agg2. We can combine global-agg1" +
- " and local-agg2 into a single operator (we call it incremental agg because " +
- "it receives incremental accumulators and output incremental results). " +
- "In this way, we can reduce some state overhead and resources. Default is enabled.")
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala
index 115a70cdeac..ece564930bd 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRelOptUtil.scala
@@ -22,7 +22,6 @@ import org.apache.flink.table.planner.analyze.PlanAdvice
import org.apache.flink.table.planner.calcite.{FlinkPlannerImpl, FlinkTypeFactory}
import org.apache.flink.table.planner.hint.FlinkHints
import org.apache.flink.table.planner.plan.`trait`.{MiniBatchInterval, MiniBatchMode}
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.calcite.config.NullCollation
import org.apache.calcite.plan.RelOptUtil
@@ -31,7 +30,7 @@ import org.apache.calcite.rel.RelFieldCollation.{Direction, NullDirection}
import org.apache.calcite.rel.core.TableScan
import org.apache.calcite.rel.hint.{Hintable, HintStrategyTable, RelHint}
import org.apache.calcite.rel.logical.{LogicalFilter, LogicalJoin, LogicalProject}
-import org.apache.calcite.rex.{RexBuilder, RexCall, RexInputRef, RexLiteral, RexNode, RexUtil, RexVisitorImpl}
+import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.SqlTypeName._
import org.apache.calcite.sql.SqlExplainLevel
import org.apache.calcite.sql.SqlKind._
@@ -270,11 +269,6 @@ object FlinkRelOptUtil {
new RelFieldCollation(fieldIndex, direction, nullDirection)
}
- /** Get max cnf node limit by context of rel */
- def getMaxCnfNodeCount(rel: RelNode): Int = {
- unwrapTableConfig(rel).get(FlinkRexUtil.TABLE_OPTIMIZER_CNF_NODES_LIMIT)
- }
-
/**
* Gets values of [[RexLiteral]] by its broad type.
*
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala
index 43888757385..82221384804 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala
@@ -17,9 +17,6 @@
*/
package org.apache.flink.table.planner.plan.utils
-import org.apache.flink.annotation.Experimental
-import org.apache.flink.configuration.ConfigOption
-import org.apache.flink.configuration.ConfigOptions.key
import org.apache.flink.table.planner.functions.sql.SqlTryCastFunction
import org.apache.flink.table.planner.plan.nodes.calcite.{LegacySink, Sink}
import org.apache.flink.table.planner.plan.optimize.RelNodeBlock
@@ -33,7 +30,7 @@ import org.apache.calcite.avatica.util.ByteString
import org.apache.calcite.plan.{RelOptPredicateList, RelOptUtil}
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.{Calc, Filter, Project, TableScan, Values}
+import org.apache.calcite.rel.core._
import org.apache.calcite.rel.logical.LogicalUnion
import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.SqlTypeName
@@ -54,31 +51,9 @@ import scala.collection.mutable
/** Utility methods concerning [[RexNode]]. */
object FlinkRexUtil {
- /** This configuration will be removed in Flink 2.0. */
- @Deprecated
- @Experimental
- val TABLE_OPTIMIZER_CNF_NODES_LIMIT: ConfigOption[Integer] =
- key("table.optimizer.cnf-nodes-limit")
- .intType()
- .defaultValue(Integer.valueOf(-1))
- .withDescription(
- "When converting to conjunctive normal form (CNF, like '(a AND b) OR" +
- " c' will be converted to '(a OR c) AND (b OR c)'), fail if the expression exceeds " +
- "this threshold; (e.g. predicate in TPC-DS q41.sql will be converted to hundreds of " +
- "thousands of CNF nodes.) the threshold is expressed in terms of number of nodes " +
- "(only count RexCall node, including leaves and interior nodes). " +
- "Negative number to use the default threshold: double of number of nodes.")
-
/**
- * Similar to [[RexUtil#toCnf(RexBuilder, Int, RexNode)]]; it lets you specify a threshold in the
- * number of nodes that can be created out of the conversion. however, if the threshold is a
- * negative number, this method will give a default threshold value that is double of the number
- * of RexCall in the given node.
- *
- * <p>If the number of resulting RexCalls exceeds that threshold, stops conversion and returns the
- * original expression.
- *
- * <p>Leaf nodes(e.g. RexInputRef) in the expression do not count towards the threshold.
+ * Converts an expression to conjunctive normal form (CNF). See more at
+ * [[RexUtil#toCnf(RexBuilder, RexNode)]].
*
* <p>We strongly discourage use the [[RexUtil#toCnf(RexBuilder, RexNode)]] and
* [[RexUtil#toCnf(RexBuilder, Int, RexNode)]], because there are many bad case when using
@@ -86,12 +61,8 @@ object FlinkRexUtil {
* to extremely complex expression (including 736450 RexCalls); and we can not give an appropriate
* value for `maxCnfNodeCount` when using [[RexUtil#toCnf(RexBuilder, Int, RexNode)]].
*/
- def toCnf(rexBuilder: RexBuilder, maxCnfNodeCount: Int, rex: RexNode): RexNode = {
- val maxCnfNodeCnt = if (maxCnfNodeCount < 0) {
- getNumberOfRexCall(rex) * 2
- } else {
- maxCnfNodeCount
- }
+ def toCnf(rexBuilder: RexBuilder, rex: RexNode): RexNode = {
+ val maxCnfNodeCnt = getNumberOfRexCall(rex) * 2
new CnfHelper(rexBuilder, maxCnfNodeCnt).toCnf(rex)
}
@@ -667,7 +638,6 @@ object FlinkRexUtil {
rel: RelNode,
rexBuilder: RexBuilder): (Array[RexNode], Array[RexNode]) = {
val context = ShortcutUtils.unwrapContext(rel)
- val maxCnfNodeCount = FlinkRelOptUtil.getMaxCnfNodeCount(rel);
val converter =
new RexNodeToExpressionConverter(
rexBuilder,
@@ -676,11 +646,7 @@ object FlinkRexUtil {
context.getCatalogManager,
Some(rel.getRowType));
- RexNodeExtractor.extractConjunctiveConditions(
- filterExpression,
- maxCnfNodeCount,
- rexBuilder,
- converter);
+ RexNodeExtractor.extractConjunctiveConditions(filterExpression, rexBuilder, converter);
}
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/IntervalJoinUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/IntervalJoinUtil.scala
index 07fc3452e57..6f1803badfb 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/IntervalJoinUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/IntervalJoinUtil.scala
@@ -78,10 +78,7 @@ object IntervalJoinUtil {
classLoader: ClassLoader): (Option[WindowBounds], Option[RexNode]) = {
// Converts the condition to conjunctive normal form (CNF)
- val cnfCondition = FlinkRexUtil.toCnf(
- rexBuilder,
- tableConfig.get(FlinkRexUtil.TABLE_OPTIMIZER_CNF_NODES_LIMIT),
- predicate)
+ val cnfCondition = FlinkRexUtil.toCnf(rexBuilder, predicate)
// split the condition into time predicates and other predicates
// We need two range predicates or an equality predicate for a properly bounded window join.
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala
index e4719f5d6c9..55325b70f0f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RankUtil.scala
@@ -23,12 +23,12 @@ import org.apache.flink.table.planner.codegen.ExpressionReducer
import org.apache.flink.table.planner.plan.nodes.calcite.Rank
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalRank
import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeduplicate
-import org.apache.flink.table.runtime.operators.rank.{ConstantRankRange, ConstantRankRangeWithoutEnd, RankRange, RankType, VariableRankRange}
+import org.apache.flink.table.runtime.operators.rank._
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.RelCollation
-import org.apache.calcite.rex.{RexBuilder, RexCall, RexInputRef, RexLiteral, RexNode, RexUtil}
+import org.apache.calcite.rex._
import org.apache.calcite.sql.SqlKind
import java.util
@@ -77,9 +77,7 @@ object RankUtil {
tableConfig: TableConfig,
classLoader: ClassLoader): (Option[RankRange], Option[RexNode]) = {
val predicate = FlinkRexUtil.expandSearch(rexBuilder, oriPred)
- // Converts the condition to conjunctive normal form (CNF)
- val cnfNodeCount = tableConfig.get(FlinkRexUtil.TABLE_OPTIMIZER_CNF_NODES_LIMIT)
- val cnfCondition = FlinkRexUtil.toCnf(rexBuilder, cnfNodeCount, predicate)
+ val cnfCondition = FlinkRexUtil.toCnf(rexBuilder, predicate)
// split the condition into sort limit condition and other condition
val (limitPreds: Seq[LimitPredicate], otherPreds: Seq[RexNode]) = cnfCondition match {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
index d27e34a664e..6a81f158bfb 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
@@ -19,7 +19,7 @@ package org.apache.flink.table.planner.plan.utils
import org.apache.flink.annotation.VisibleForTesting
import org.apache.flink.table.api.TableException
-import org.apache.flink.table.catalog.{CatalogManager, ContextResolvedFunction, FunctionCatalog, FunctionLookup, UnresolvedIdentifier}
+import org.apache.flink.table.catalog.{CatalogManager, ContextResolvedFunction, FunctionCatalog, UnresolvedIdentifier}
import org.apache.flink.table.data.conversion.{DayTimeIntervalDurationConverter, YearMonthIntervalPeriodConverter}
import org.apache.flink.table.data.util.DataFormatConverters.{LocalDateConverter, LocalTimeConverter}
import org.apache.flink.table.expressions._
@@ -37,7 +37,6 @@ import org.apache.flink.table.types.logical.YearMonthIntervalType
import org.apache.flink.table.types.utils.TypeConversions
import org.apache.flink.util.Preconditions
-import org.apache.calcite.avatica.util.ByteString
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rex._
@@ -49,9 +48,9 @@ import java.time.{ZoneId, ZoneOffset}
import java.util
import java.util.{Collections, List => JList}
-import scala.collection.{mutable, JavaConverters}
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
+import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.util.{Failure, Success, Try}
@@ -107,7 +106,6 @@ object RexNodeExtractor extends Logging {
*/
def extractConjunctiveConditions(
expr: RexNode,
- maxCnfNodeCount: Int,
inputFieldNames: JList[String],
rexBuilder: RexBuilder,
functionCatalog: FunctionCatalog,
@@ -116,7 +114,7 @@ object RexNodeExtractor extends Logging {
val converter =
new RexNodeToExpressionConverter(rexBuilder, inputNames, functionCatalog, catalogManager)
val (convertibleRexNodes, unconvertedRexNodes) =
- extractConjunctiveConditions(expr, maxCnfNodeCount, rexBuilder, converter)
+ extractConjunctiveConditions(expr, rexBuilder, converter)
val convertedExpressions = convertibleRexNodes.map(_.accept(converter).get)
(convertedExpressions.toArray, unconvertedRexNodes)
}
@@ -135,7 +133,6 @@ object RexNodeExtractor extends Logging {
*/
def extractConjunctiveConditions(
expr: RexNode,
- maxCnfNodeCount: Int,
rexBuilder: RexBuilder,
converter: RexNodeToExpressionConverter): (Array[RexNode], Array[RexNode]) = {
// converts the expanded expression to conjunctive normal form,
@@ -147,7 +144,7 @@ object RexNodeExtractor extends Logging {
} else {
expr
}
- val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, rewrite)
+ val cnf = FlinkRexUtil.toCnf(rexBuilder, rewrite)
// converts the cnf condition to a list of AND conditions
val conjunctions = RelOptUtil.conjunctions(cnf)
@@ -166,16 +163,11 @@ object RexNodeExtractor extends Logging {
@VisibleForTesting
def extractPartitionPredicates(
expr: RexNode,
- maxCnfNodeCount: Int,
inputFieldNames: Array[String],
rexBuilder: RexBuilder,
partitionFieldNames: Array[String]): (RexNode, RexNode) = {
- val (partitionPredicates, nonPartitionPredicates) = extractPartitionPredicateList(
- expr,
- maxCnfNodeCount,
- inputFieldNames,
- rexBuilder,
- partitionFieldNames)
+ val (partitionPredicates, nonPartitionPredicates) =
+ extractPartitionPredicateList(expr, inputFieldNames, rexBuilder, partitionFieldNames)
val partitionPredicate = RexUtil.composeConjunction(rexBuilder, partitionPredicates)
val nonPartitionPredicate = RexUtil.composeConjunction(rexBuilder, nonPartitionPredicates)
(partitionPredicate, nonPartitionPredicate)
@@ -197,13 +189,12 @@ object RexNodeExtractor extends Logging {
*/
def extractPartitionPredicateList(
expr: RexNode,
- maxCnfNodeCount: Int,
inputFieldNames: Array[String],
rexBuilder: RexBuilder,
partitionFieldNames: Array[String]): (Seq[RexNode], Seq[RexNode]) = {
// converts the expanded expression to conjunctive normal form,
// like "(a AND b) OR c" will be converted to "(a OR c) AND (b OR c)"
- val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, expr)
+ val cnf = FlinkRexUtil.toCnf(rexBuilder, expr)
// converts the cnf condition to a list of AND conditions
val conjunctions = RelOptUtil.conjunctions(cnf)
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml
index 50be25b3c72..958420904ce 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml
@@ -139,7 +139,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f1=[$3], f=[$4], cnt=[$5])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f1, f, cnt], rightSorted=[true])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[EXPR$0 AS f1, f, cnt])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f1, f, cnt])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[EXPR$0 AS f1, f, cnt])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) AS cnt])\n [...]
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
@@ -175,7 +175,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f=[$3], f1=[$4])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f, f1], rightSorted=[true])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f, EXPR$0 AS f1])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f, f1])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f, EXPR$0 AS f1])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) AS cnt])\n + [...]
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
@@ -252,7 +252,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a1=[$6], b1=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(a = d1)], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1], leftSorted=[true])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(a = a1)], select=[a, b, c, d, e, f, a1, b1, c1], leftSorted=[true])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a]])\n: : +- [#4] Exchange(distribution=[hash[d]])\n: +- [#2] Exchange(distribution=[hash[ [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(a = d1)], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(a = a1)], select=[a, b, c, d, e, f, a1, b1, c1])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a]])\n: : +- [#4] Exchange(distribution=[hash[d]])\n: +- [#2] Exchange(distribution=[hash[a1]])\n+- [#1] Exchange(distribution=[ [...]
:- Exchange(distribution=[hash[d1]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, t2, source: [TestTableSource(d1, e1, f1)]]], fields=[d1, e1, f1])
:- Exchange(distribution=[hash[a1]])
@@ -287,7 +287,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a1=[$6], b1=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(a1 = d1)], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1], leftSorted=[true])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(d = a1)], select=[a, b, c, d, e, f, a1, b1, c1], leftSorted=[true])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a]])\n: : +- [#4] Exchange(distribution=[hash[d]])\n: +- [#2] Exchange(distribution=[hash [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(a1 = d1)], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(d = a1)], select=[a, b, c, d, e, f, a1, b1, c1])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a]])\n: : +- [#4] Exchange(distribution=[hash[d]])\n: +- [#2] Exchange(distribution=[hash[a1]])\n+- [#1] Exchange(distribution= [...]
:- Exchange(distribution=[hash[d1]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, t2, source: [TestTableSource(d1, e1, f1)]]], fields=[d1, e1, f1])
:- Exchange(distribution=[hash[a1]])
@@ -374,12 +374,12 @@ LogicalProject(id=[$0], tb2_ids=[$2], tb3_ids=[$3], name=[$4], tb2_names=[$6], t
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[id, tb2_ids, tb3_ids, name, tb2_names, tb3_names, name0, name1])
-+- MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(key = id0)], select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb3_names, name0, id0, name1], leftSorted=[true])\n:- Calc(select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb3_names, name0])\n: +- SortMergeJoin(joinType=[LeftOuterJoin], where=[(key = id0)], select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb3_names, id0, name0])\n: :- [#2] Exchange(distribution=[hash[key]])\n: +- [#3] Exchange(dis [...]
++- MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(key = id0)], select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb3_names, name0, id0, name1])\n:- Calc(select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb3_names, name0])\n: +- SortMergeJoin(joinType=[LeftOuterJoin], where=[(key = id0)], select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb3_names, id0, name0])\n: :- [#2] Exchange(distribution=[hash[key]])\n: +- [#3] Exchange(distribution=[hash[id] [...]
:- Exchange(distribution=[hash[id]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, tb5, source: [TestTableSource(id, name)]]], fields=[id, name])
:- Exchange(distribution=[hash[key]])
: +- Calc(select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb3_names])
- : +- MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(id = tb1_id)], select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb1_id, tb3_names], leftSorted=[true], rightSorted=[true])\n:- Calc(select=[id, key, tb2_ids, tb3_ids, name, tb2_names])\n: +- SortMergeJoin(joinType=[LeftOuterJoin], where=[(id = tb1_id)], select=[id, key, tb2_ids, tb3_ids, name, tb1_id, tb2_names], rightSorted=[true])\n: :- [#1] Exchange(distribution=[hash[id]])\n: +- SortAggre [...]
+ : +- MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[(id = tb1_id)], select=[id, key, tb2_ids, tb3_ids, name, tb2_names, tb1_id, tb3_names])\n:- Calc(select=[id, key, tb2_ids, tb3_ids, name, tb2_names])\n: +- SortMergeJoin(joinType=[LeftOuterJoin], where=[(id = tb1_id)], select=[id, key, tb2_ids, tb3_ids, name, tb1_id, tb2_names])\n: :- [#1] Exchange(distribution=[hash[id]])\n: +- SortAggregate(isMerge=[true], groupBy=[tb1_id], select=[tb1_id, Fina [...]
: :- Exchange(distribution=[hash[id]])
: : +- LegacyTableSourceScan(table=[[default_catalog, default_database, tb1, source: [TestTableSource(id, key, tb2_ids, tb3_ids, name)]]], fields=[id, key, tb2_ids, tb3_ids, name])(reuse_id=[1])
: :- Exchange(distribution=[hash[tb1_id]])
@@ -434,7 +434,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a1=[$6], b1=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[((a = d1) AND (b = e1))], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1], leftSorted=[true])\n:- SortMergeJoin(joinType=[InnerJoin], where=[((a = a1) AND (b = b1))], select=[a, b, c, d, e, f, a1, b1, c1], leftSorted=[true])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a, b]])\n: : +- [#4] Exchange(distribution=[has [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[((a = d1) AND (b = e1))], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1])\n:- SortMergeJoin(joinType=[InnerJoin], where=[((a = a1) AND (b = b1))], select=[a, b, c, d, e, f, a1, b1, c1])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a, b]])\n: : +- [#4] Exchange(distribution=[hash[d, e]])\n: +- [#2] Exchange(distrib [...]
:- Exchange(distribution=[hash[d1, e1]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, t2, source: [TestTableSource(d1, e1, f1)]]], fields=[d1, e1, f1])
:- Exchange(distribution=[hash[a1, b1]])
@@ -469,7 +469,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a1=[$6], b1=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[((a1 = d1) AND (b1 = e1))], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1], leftSorted=[true])\n:- SortMergeJoin(joinType=[InnerJoin], where=[((d = a1) AND (e = b1))], select=[a, b, c, d, e, f, a1, b1, c1], leftSorted=[true])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a, b]])\n: : +- [#4] Exchange(distribution=[h [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[LeftOuterJoin], where=[((a1 = d1) AND (b1 = e1))], select=[a, b, c, d, e, f, a1, b1, c1, d1, e1, f1])\n:- SortMergeJoin(joinType=[InnerJoin], where=[((d = a1) AND (e = b1))], select=[a, b, c, d, e, f, a1, b1, c1])\n: :- SortMergeJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, c, d, e, f])\n: : :- [#3] Exchange(distribution=[hash[a, b]])\n: : +- [#4] Exchange(distribution=[hash[d, e]])\n: +- [#2] Exchange(distr [...]
:- Exchange(distribution=[hash[d1, e1]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, t2, source: [TestTableSource(d1, e1, f1)]]], fields=[d1, e1, f1])
:- Exchange(distribution=[hash[a1, b1]])
@@ -535,7 +535,7 @@ LogicalSort(sort0=[$0], dir0=[ASC-nulls-first])
<Resource name="optimized rel plan">
<![CDATA[
Sort(orderBy=[a ASC])
-+- Exchange(distribution=[range[a ASC]])
++- Exchange(distribution=[single])
+- HashAggregate(isMerge=[true], groupBy=[a, b], select=[a, b, Final_COUNT(count$0) AS cnt])
+- Exchange(distribution=[hash[a, b]])
+- LocalHashAggregate(groupBy=[a, b], select=[a, b, Partial_COUNT(c) AS count$0])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
index fba75b39f7f..97a174f1905 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
@@ -36,12 +36,12 @@ LogicalProject(EXPR$0=[$1])
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[EXPR$0])
-+- HashAggregate(isMerge=[false], groupBy=[c], select=[c, COUNT(*) AS EXPR$0])
- +- Exchange(distribution=[keep_input_as_is[hash[c]]])
- +- Calc(select=[c])
- +- Exchange(distribution=[keep_input_as_is[hash[a]]])
++- HashAggregate(isMerge=[true], groupBy=[c], select=[c, Final_COUNT(count1$0) AS EXPR$0])
+ +- Exchange(distribution=[hash[c]])
+ +- LocalHashAggregate(groupBy=[c], select=[c, Partial_COUNT(*) AS count1$0])
+ +- Calc(select=[c])
+- HashAggregate(isMerge=[true], groupBy=[a, c], select=[a, c])
- +- Exchange(distribution=[hash[c]])
+ +- Exchange(distribution=[hash[a, c]])
+- LocalHashAggregate(groupBy=[a, c], select=[a, c])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
@@ -526,13 +526,15 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], cnt=[$4])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(readOrder=[1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = cnt))], select=[a, b, c, d, cnt], build=[right])\n:- [#1] Exchange(distribution=[hash[a]])\n+- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS cnt])\n +- [#2] Exchange(distribution=[hash[d]])\n])
-:- Exchange(distribution=[hash[a]])
+HashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = cnt))], select=[a, b, c, d, cnt], build=[right])
+:- Exchange(distribution=[hash[a, b]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-+- Exchange(distribution=[hash[d]])
- +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(f) AS count$0])
- +- Calc(select=[d, f])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
++- Exchange(distribution=[hash[d, cnt]])
+ +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS cnt])
+ +- Exchange(distribution=[hash[d]])
+ +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(f) AS count$0])
+ +- Calc(select=[d, f])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
</TestCase>
@@ -674,14 +676,13 @@ LogicalProject(sum_b=[$2], avg_b=[/(CASE(>(COUNT($2) OVER (PARTITION BY $1), 0),
Calc(select=[sum_b, (CASE((w0$o0 > 0), w0$o1, null:BIGINT) / w0$o0) AS avg_b, w1$o0 AS rn, c])
+- OverAggregate(partitionBy=[c], window#0=[COUNT(sum_b) AS w0$o0, $SUM0(sum_b) AS w0$o1 RANG BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING], window#1=[RANK(*) AS w1$o0 RANG BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], select=[c, sum_b, w0$o0, w0$o1, w1$o0])
+- Exchange(distribution=[forward])
- +- Calc(select=[c, sum_b])
- +- Exchange(distribution=[forward])
- +- Sort(orderBy=[c ASC])
- +- Exchange(distribution=[forward])
- +- HashAggregate(isMerge=[true], groupBy=[a, c], select=[a, c, Final_SUM(sum$0) AS sum_b])
- +- Exchange(distribution=[hash[c]])
- +- LocalHashAggregate(groupBy=[a, c], select=[a, c, Partial_SUM(b) AS sum$0])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+ +- Sort(orderBy=[c ASC])
+ +- Exchange(distribution=[hash[c]])
+ +- Calc(select=[c, sum_b])
+ +- HashAggregate(isMerge=[true], groupBy=[a, c], select=[a, c, Final_SUM(sum$0) AS sum_b])
+ +- Exchange(distribution=[hash[a, c]])
+ +- LocalHashAggregate(groupBy=[a, c], select=[a, c, Partial_SUM(b) AS sum$0])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
@@ -742,17 +743,16 @@ LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)])
<Resource name="optimized exec plan">
<![CDATA[
HashAggregate(isMerge=[false], groupBy=[a], select=[a, SUM(b) AS EXPR$1])
-+- Exchange(distribution=[keep_input_as_is[hash[a]]])
++- Exchange(distribution=[hash[a]])
+- Calc(select=[a, b])
- +- Exchange(distribution=[keep_input_as_is[hash[a]]])
- +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[true], select=[a, b, c])
- +- Exchange(distribution=[forward])
- +- Sort(orderBy=[a ASC, c ASC, b ASC])
- +- Exchange(distribution=[hash[a]])
- +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[false], select=[a, b, c])
- +- Exchange(distribution=[forward])
- +- Sort(orderBy=[a ASC, c ASC, b ASC])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+ +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[true], select=[a, b, c])
+ +- Exchange(distribution=[forward])
+ +- Sort(orderBy=[a ASC, c ASC, b ASC])
+ +- Exchange(distribution=[hash[a, c]])
+ +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[false], select=[a, b, c])
+ +- Exchange(distribution=[forward])
+ +- Sort(orderBy=[a ASC, c ASC, b ASC])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
@@ -1049,7 +1049,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a0=[$6], b0=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0], leftSorted=[true], rightSorted=[true])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: :- [#1] Exchange(distribution=[hash[a]])\n: +- [#2] Exchange(distribution=[hash[d]])\n+- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n :- [#3] Exchange(distribution=[hash[a]])\n +- [#4] Exchange [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: :- [#1] Exchange(distribution=[hash[a]])\n: +- [#2] Exchange(distribution=[hash[d]])\n+- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n :- [#3] Exchange(distribution=[hash[a]])\n +- [#4] Exchange(distribution=[hash[d]])\n])
:- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, b, c], where=[LIKE(c, 'He%')])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
@@ -1137,7 +1137,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a0=[$6], b0=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0], leftSorted=[true])\n:- SortMergeJoin(joinType=[LeftOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: :- [#2] Exchange(distribution=[hash[a]])\n: +- [#3] Exchange(distribution=[hash[d]])\n+- [#1] Exchange(distribution=[hash[d]])\n])
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0])\n:- SortMergeJoin(joinType=[LeftOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n: :- [#2] Exchange(distribution=[hash[a]])\n: +- [#3] Exchange(distribution=[hash[d]])\n+- [#1] Exchange(distribution=[hash[d]])\n])
:- Exchange(distribution=[hash[d]])
: +- SortMergeJoin(joinType=[LeftOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])
: :- Exchange(distribution=[hash[a]])
@@ -1181,7 +1181,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a0=[$6], b0=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0], rightSorted=[true])\n:- [#1] Exchange(distribution=[hash[a]])\n+- SortMergeJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n :- [#2] Exchange(distribution=[hash[a]])\n +- [#3] Exchange(distribution=[hash[d]])\n])
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0])\n:- [#1] Exchange(distribution=[hash[a]])\n+- SortMergeJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n :- [#2] Exchange(distribution=[hash[a]])\n +- [#3] Exchange(distribution=[hash[d]])\n])
:- Exchange(distribution=[hash[a]])
: +- SortMergeJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])
: :- Exchange(distribution=[hash[a]])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
index 44b5a58b63f..1b0be44aa16 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
@@ -48,9 +48,10 @@ LogicalSort(sort0=[$0], dir0=[DESC-nulls-last])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Sort(orderBy=[a DESC])
-+- Exchange(distribution=[range[a DESC]])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+SortLimit(orderBy=[a DESC], offset=[0], fetch=[200], global=[true])
++- Exchange(distribution=[single])
+ +- SortLimit(orderBy=[a DESC], offset=[0], fetch=[200], global=[false])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
@@ -87,7 +88,7 @@ LogicalSort(sort0=[$0], dir0=[DESC-nulls-last])
<Resource name="optimized rel plan">
<![CDATA[
Sort(orderBy=[a DESC])
-+- Exchange(distribution=[range[a DESC]])
++- Exchange(distribution=[single])
+- 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/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
index e0fbe17efbd..6cf86314772 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
@@ -1332,7 +1332,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a0=[$6], b0=[$7],
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0], leftSorted=[true], rightSorted=[true])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])(reuse_id=[1])\n: :- [#1] Exchange(distribution=[hash[a]])\n: +- [#2] Exchange(distribution=[hash[d]])\n+- Reused(reference_id=[1])\n])
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0])\n:- SortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, b, c, d, e, f])(reuse_id=[1])\n: :- [#1] Exchange(distribution=[hash[a]])\n: +- [#2] Exchange(distribution=[hash[d]])\n+- Reused(reference_id=[1])\n])
:- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, b, c], where=[LIKE(c, 'He%')])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.xml
index 2c5b94ea524..0d00419daaf 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.xml
@@ -31,14 +31,16 @@ LogicalAggregate(group=[{0}], EXPR$1=[MAX($1)], EXPR$2=[MAX($1) FILTER $2], EXPR
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-HashAggregate(isMerge=[false], groupBy=[d], select=[d, MIN(EXPR$1) FILTER $g_15 AS EXPR$1, MIN(EXPR$2) FILTER $g_15 AS EXPR$2, COUNT(c) FILTER $g_7 AS EXPR$3, COUNT(c) FILTER $g_3 AS EXPR$4, COUNT(b) FILTER $g_12 AS EXPR$5])
-+- Calc(select=[d, c, b, EXPR$1, EXPR$2, AND(=(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 3), IS TRUE($f4)) AS $g_3, =(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 7) AS $g_7, AND(=(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 12), IS TRUE($f6)) AS $g_12, =(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 15) AS $g_15])
- +- HashAggregate(isMerge=[true], groupBy=[d, c, $f4, b, $f6, $e], select=[d, c, $f4, b, $f6, $e, Final_MAX(max$0) AS EXPR$1, Final_MAX(max$1) AS EXPR$2])
- +- Exchange(distribution=[hash[d]])
- +- LocalHashAggregate(groupBy=[d, c, $f4, b, $f6, $e], select=[d, c, $f4, b, $f6, $e, Partial_MAX(e) AS max$0, Partial_MAX(e) FILTER $f2 AS max$1])
- +- Expand(projects=[{d, e, $f2, c, $f4, null AS b, null AS $f6, 3 AS $e}, {d, e, $f2, c, null AS $f4, null AS b, null AS $f6, 7 AS $e}, {d, e, $f2, null AS c, null AS $f4, b, $f6, 12 AS $e}, {d, e, $f2, null AS c, null AS $f4, null AS b, null AS $f6, 15 AS $e}])
- +- Calc(select=[d, e, IS TRUE(<(a, 10)) AS $f2, c, IS TRUE(>(a, 5)) AS $f4, b, IS TRUE(>(b, 3)) AS $f6])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, z, source: [TestTableSource(a, b, c, d, e)]]], fields=[a, b, c, d, e])
+HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_MIN(min$0) AS EXPR$1, Final_MIN(min$1) AS EXPR$2, Final_COUNT(count$2) AS EXPR$3, Final_COUNT(count$3) AS EXPR$4, Final_COUNT(count$4) AS EXPR$5])
++- Exchange(distribution=[hash[d]])
+ +- LocalHashAggregate(groupBy=[d], select=[d, Partial_MIN(EXPR$1) FILTER $g_15 AS min$0, Partial_MIN(EXPR$2) FILTER $g_15 AS min$1, Partial_COUNT(c) FILTER $g_7 AS count$2, Partial_COUNT(c) FILTER $g_3 AS count$3, Partial_COUNT(b) FILTER $g_12 AS count$4])
+ +- Calc(select=[d, c, b, EXPR$1, EXPR$2, AND(=(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 3), IS TRUE($f4)) AS $g_3, =(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 7) AS $g_7, AND(=(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 12), IS TRUE($f6)) AS $g_12, =(CASE(=($e, 3), 3, =($e, 7), 7, =($e, 12), 12, 15), 15) AS $g_15])
+ +- HashAggregate(isMerge=[true], groupBy=[d, c, $f4, b, $f6, $e], select=[d, c, $f4, b, $f6, $e, Final_MAX(max$0) AS EXPR$1, Final_MAX(max$1) AS EXPR$2])
+ +- Exchange(distribution=[hash[d, c, $f4, b, $f6, $e]])
+ +- LocalHashAggregate(groupBy=[d, c, $f4, b, $f6, $e], select=[d, c, $f4, b, $f6, $e, Partial_MAX(e) AS max$0, Partial_MAX(e) FILTER $f2 AS max$1])
+ +- Expand(projects=[{d, e, $f2, c, $f4, null AS b, null AS $f6, 3 AS $e}, {d, e, $f2, c, null AS $f4, null AS b, null AS $f6, 7 AS $e}, {d, e, $f2, null AS c, null AS $f4, b, $f6, 12 AS $e}, {d, e, $f2, null AS c, null AS $f4, null AS b, null AS $f6, 15 AS $e}])
+ +- Calc(select=[d, e, IS TRUE(<(a, 10)) AS $f2, c, IS TRUE(>(a, 5)) AS $f4, b, IS TRUE(>(b, 3)) AS $f6])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, z, source: [TestTableSource(a, b, c, d, e)]]], fields=[a, b, c, d, e])
]]>
</Resource>
</TestCase>
@@ -98,14 +100,16 @@ LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-HashAggregate(isMerge=[false], groupBy=[a], select=[a, SUM(b) AS EXPR$1])
-+- Calc(select=[a, b])
- +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[true], select=[a, b, c])
- +- Sort(orderBy=[a ASC, c ASC, b ASC])
- +- Exchange(distribution=[hash[a]])
- +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[false], select=[a, b, c])
- +- Sort(orderBy=[a ASC, c ASC, b ASC])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_SUM(sum$0) AS EXPR$1])
++- Exchange(distribution=[hash[a]])
+ +- LocalHashAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0])
+ +- Calc(select=[a, b])
+ +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[true], select=[a, b, c])
+ +- Sort(orderBy=[a ASC, c ASC, b ASC])
+ +- Exchange(distribution=[hash[a, c]])
+ +- Rank(rankType=[RANK], rankRange=[rankStart=1, rankEnd=10], partitionBy=[a, c], orderBy=[b ASC], global=[false], select=[a, b, c])
+ +- Sort(orderBy=[a ASC, c ASC, b ASC])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala
index e2359f6b98a..2514a7a2c4f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala
@@ -21,8 +21,6 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.table.api.Types
import org.apache.flink.table.api.config.ExecutionConfigOptions
import org.apache.flink.table.plan.stats.TableStats
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortMergeJoinRule
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED
import org.apache.flink.table.planner.plan.stats.FlinkStatistic
import org.apache.flink.table.planner.runtime.utils.JavaUserDefinedTableFunctions.StringSplit
import org.apache.flink.table.planner.utils.{TableFunc1, TableTestBase}
@@ -58,10 +56,6 @@ class RemoveCollationTest extends TableTestBase {
Array("d1", "e1", "f1"),
FlinkStatistic.builder().tableStats(new TableStats(100L)).build()
)
-
- util.tableEnv.getConfig.set(
- BatchPhysicalSortMergeJoinRule.TABLE_OPTIMIZER_SMJ_REMOVE_SORT_ENABLED,
- Boolean.box(true))
}
@Test
@@ -107,7 +101,6 @@ class RemoveCollationTest extends TableTestBase {
@Test
def testRemoveCollation_Sort(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT a, b, COUNT(c) AS cnt FROM x GROUP BY a, b)
@@ -120,7 +113,6 @@ class RemoveCollationTest extends TableTestBase {
@Test
def testRemoveCollation_Aggregate_3(): Unit = {
util.tableEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashAgg")
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT * FROM x ORDER BY a, b)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala
index 5c1e6e8614f..e8cdff6ebc0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala
@@ -21,7 +21,6 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.table.api.Types
import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions}
import org.apache.flink.table.plan.stats.TableStats
-import org.apache.flink.table.planner.plan.rules.physical.batch.{BatchPhysicalJoinRuleBase, BatchPhysicalSortMergeJoinRule}
import org.apache.flink.table.planner.plan.stats.FlinkStatistic
import org.apache.flink.table.planner.utils.{TableFunc1, TableTestBase}
@@ -88,9 +87,6 @@ class RemoveShuffleTest extends TableTestBase {
util.tableEnv.getConfig.set(
ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS,
"NestedLoopJoin,SortMergeJoin,SortAgg")
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(true))
// push down HashExchange[c] into HashAgg
val sqlQuery =
"""
@@ -110,9 +106,6 @@ class RemoveShuffleTest extends TableTestBase {
util.tableEnv.getConfig.set(
ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS,
"NestedLoopJoin,SortMergeJoin,SortAgg")
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(true))
// push down HashExchange[c] into HashAgg
val sqlQuery =
"""
@@ -222,9 +215,6 @@ class RemoveShuffleTest extends TableTestBase {
def testRemoveHashShuffle_SortMergeJoin(): Unit = {
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin,NestedLoopJoin")
- util.tableEnv.getConfig.set(
- BatchPhysicalSortMergeJoinRule.TABLE_OPTIMIZER_SMJ_REMOVE_SORT_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT * FROM x, y WHERE a = d AND c LIKE 'He%')
@@ -237,9 +227,6 @@ class RemoveShuffleTest extends TableTestBase {
def testRemoveHashShuffle_SortMergeJoin_LOJ(): Unit = {
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin,NestedLoopJoin")
- util.tableEnv.getConfig.set(
- BatchPhysicalSortMergeJoinRule.TABLE_OPTIMIZER_SMJ_REMOVE_SORT_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT * FROM x left join (SELECT * FROM y WHERE e = 2) r on a = d)
@@ -252,9 +239,6 @@ class RemoveShuffleTest extends TableTestBase {
def testRemoveHashShuffle_SortMergeJoin_ROJ(): Unit = {
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin,NestedLoopJoin")
- util.tableEnv.getConfig.set(
- BatchPhysicalSortMergeJoinRule.TABLE_OPTIMIZER_SMJ_REMOVE_SORT_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT * FROM x right join (SELECT * FROM y WHERE e = 2) r on a = d)
@@ -383,9 +367,6 @@ class RemoveShuffleTest extends TableTestBase {
// disable BroadcastHashJoin
util.tableEnv.getConfig
.set(OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD, Long.box(-1))
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT d, count(f) as cnt FROM y GROUP BY d)
@@ -433,9 +414,6 @@ class RemoveShuffleTest extends TableTestBase {
@Test
def testRemoveHashShuffle_Rank_PartialKey1(): Unit = {
util.tableEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|SELECT a, SUM(b) FROM (
@@ -450,9 +428,6 @@ class RemoveShuffleTest extends TableTestBase {
@Test
def testRemoveHashShuffle_Rank_PartialKey2(): Unit = {
util.tableEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(false))
val sqlQuery =
"""
|SELECT * FROM (
@@ -467,9 +442,6 @@ class RemoveShuffleTest extends TableTestBase {
@Test
def testRemoveHashShuffle_Rank_PartialKey3(): Unit = {
util.tableEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|SELECT * FROM (
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortLimitTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortLimitTest.scala
index 80fdc14559f..9d22095e799 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortLimitTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortLimitTest.scala
@@ -19,7 +19,6 @@ package org.apache.flink.table.planner.plan.batch.sql
import org.apache.flink.table.api._
import org.apache.flink.table.api.config.ExecutionConfigOptions
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED
import org.apache.flink.table.planner.utils.TableTestBase
import org.junit.jupiter.api.Test
@@ -33,61 +32,51 @@ class SortLimitTest extends TableTestBase {
@Test
def testNonRangeSortWithoutOffset(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC LIMIT 5")
}
@Test
def testNonRangeSortWithLimit0(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC LIMIT 0")
}
@Test
def testNonRangeSortOnlyWithOffset(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC OFFSET 10 ROWS")
}
@Test
def testNoneRangeSortWithOffsetLimit(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC, b LIMIT 10 OFFSET 1")
}
@Test
def testNoneRangeSortWithOffsetLimit0(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC, b LIMIT 0 OFFSET 1")
}
@Test
def testRangeSortOnWithoutOffset(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC LIMIT 5")
}
@Test
def testRangeSortOnWithLimit0(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC LIMIT 0")
}
@Test
def testRangeSortOnlyWithOffset(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC OFFSET 10 ROWS")
}
@Test
def testRangeSortWithOffsetLimit(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC, b LIMIT 10 OFFSET 1")
}
@Test
def testRangeSortWithOffsetLimit0(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC, b LIMIT 0 OFFSET 1")
}
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala
index ac79592b153..518d86882b9 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala
@@ -19,7 +19,6 @@ package org.apache.flink.table.planner.plan.batch.sql
import org.apache.flink.table.api._
import org.apache.flink.table.api.config.ExecutionConfigOptions
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED
import org.apache.flink.table.planner.utils.TableTestBase
import org.junit.jupiter.api.Test
@@ -31,7 +30,6 @@ class SortTest extends TableTestBase {
@Test
def testNonRangeSortOnSingleFieldWithoutForceLimit(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_SORT_DEFAULT_LIMIT, Integer.valueOf(-1))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC")
@@ -39,7 +37,6 @@ class SortTest extends TableTestBase {
@Test
def testNonRangeSortOnMultiFieldsWithoutForceLimit(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_SORT_DEFAULT_LIMIT, Integer.valueOf(-1))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC, b")
@@ -47,7 +44,6 @@ class SortTest extends TableTestBase {
@Test
def testNonRangeSortWithForceLimit(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(false))
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_SORT_DEFAULT_LIMIT, Integer.valueOf(200))
util.verifyExecPlan("SELECT * FROM MyTable ORDER BY a DESC")
@@ -55,7 +51,6 @@ class SortTest extends TableTestBase {
@Test
def testRangeSortWithoutForceLimit(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_SORT_DEFAULT_LIMIT, Integer.valueOf(-1))
// exec node does not support range sort yet, so we verify rel plan here
@@ -64,7 +59,6 @@ class SortTest extends TableTestBase {
@Test
def testRangeSortWithForceLimit(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_SORT_DEFAULT_LIMIT, Integer.valueOf(200))
// exec node does not support range sort yet, so we verify rel plan here
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala
index 3d46d4fcb85..b2d4f4cf449 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala
@@ -19,8 +19,6 @@ package org.apache.flink.table.planner.plan.batch.sql
import org.apache.flink.table.api._
import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfigOptions}
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortMergeJoinRule
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED
import org.apache.flink.table.planner.plan.utils.OperatorType
import org.apache.flink.table.planner.runtime.utils.JavaUserDefinedScalarFunctions.NonDeterministicUdf
import org.apache.flink.table.planner.runtime.utils.JavaUserDefinedTableFunctions.{NonDeterministicTableFunc, StringSplit}
@@ -220,7 +218,6 @@ class SubplanReuseTest extends TableTestBase {
@Test
def testSubplanReuseOnSort(): Unit = {
- util.tableEnv.getConfig.set(TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT c, SUM(a) a, SUM(b) b FROM x GROUP BY c ORDER BY a, b DESC)
@@ -276,9 +273,6 @@ class SubplanReuseTest extends TableTestBase {
def testSubplanReuseOnSortMergeJoin(): Unit = {
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin,NestedLoopJoin")
- util.tableEnv.getConfig.set(
- BatchPhysicalSortMergeJoinRule.TABLE_OPTIMIZER_SMJ_REMOVE_SORT_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|WITH r AS (SELECT * FROM x, y WHERE a = d AND c LIKE 'He%')
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/JoinReorderTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/JoinReorderTestBase.scala
index 02aae7ee592..16d2b8b024b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/JoinReorderTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/JoinReorderTestBase.scala
@@ -455,13 +455,13 @@ abstract class JoinReorderTestBase(isBushyJoinReorder: Boolean) extends TableTes
val types = Array[TypeInformation[_]](Types.INT, Types.LONG)
val builderA = ColumnStats.Builder
.builder()
- .setNdv(200000L)
- .setNullCount(50000L)
+ .setNdv(2000000L)
+ .setNullCount(JoinDeriveNullFilterRule.JOIN_NULL_FILTER_THRESHOLD + 1)
.setAvgLen(4.0)
.setMaxLen(4)
val builderB = ColumnStats.Builder
.builder()
- .setNdv(100000L)
+ .setNdv(1000000L)
.setNullCount(0L)
.setAvgLen(8.0)
.setMaxLen(8)
@@ -474,7 +474,7 @@ abstract class JoinReorderTestBase(isBushyJoinReorder: Boolean) extends TableTes
.builder()
.tableStats(
new TableStats(
- 500000L,
+ 5000000L,
Map(
"a6" -> builderA.build(),
"b6" -> builderB.build()
@@ -489,7 +489,7 @@ abstract class JoinReorderTestBase(isBushyJoinReorder: Boolean) extends TableTes
.builder()
.tableStats(
new TableStats(
- 500000L,
+ 5000000L,
Map(
"a7" -> builderA.build(),
"b7" -> builderB.build()
@@ -504,15 +504,13 @@ abstract class JoinReorderTestBase(isBushyJoinReorder: Boolean) extends TableTes
.builder()
.tableStats(
new TableStats(
- 500000L,
+ 5000000L,
Map(
"a8" -> builderA.build(),
"b8" -> builderB.build()
)))
.build())
- util.getTableEnv.getConfig
- .set(JoinDeriveNullFilterRule.TABLE_OPTIMIZER_JOIN_NULL_FILTER_THRESHOLD, Long.box(10000))
val sql =
s"""
|SELECT * FROM T6
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRuleTest.scala
index 802b107fd4c..7dc1d78f591 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/JoinDeriveNullFilterRuleTest.scala
@@ -48,8 +48,6 @@ class JoinDeriveNullFilterRuleTest extends TableTestBase {
.build()
)
- util.tableEnv.getConfig
- .set(JoinDeriveNullFilterRule.TABLE_OPTIMIZER_JOIN_NULL_FILTER_THRESHOLD, Long.box(2000000))
util.addTableSource(
"MyTable1",
Array[TypeInformation[_]](Types.INT, Types.LONG, Types.STRING, Types.INT, Types.LONG),
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.scala
index e2b6f45ba53..0006f210eeb 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/RemoveRedundantLocalHashAggRuleTest.scala
@@ -54,9 +54,6 @@ class RemoveRedundantLocalHashAggRuleTest extends TableTestBase {
@Test
def testRemoveRedundantLocalHashAgg_ShuffleKeyFromRank(): Unit = {
util.tableEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(true))
val sqlQuery =
"""
|SELECT a, SUM(b) FROM (
@@ -71,9 +68,6 @@ class RemoveRedundantLocalHashAggRuleTest extends TableTestBase {
@Test
def testUsingLocalAggCallFilters(): Unit = {
util.tableEnv.getConfig.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "SortAgg")
- util.tableEnv.getConfig.set(
- BatchPhysicalJoinRuleBase.TABLE_OPTIMIZER_SHUFFLE_BY_PARTIAL_KEY_ENABLED,
- Boolean.box(true))
val sqlQuery = "SELECT d, MAX(e), MAX(e) FILTER (WHERE a < 10), COUNT(DISTINCT c),\n" +
"COUNT(DISTINCT c) FILTER (WHERE a > 5), COUNT(DISTINCT b) FILTER (WHERE b > 3)\n" +
"FROM z GROUP BY d"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala
index c911b11cd71..8420517c59c 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtilTest.scala
@@ -25,7 +25,7 @@ import org.apache.calcite.rex.{RexBuilder, RexLiteral, RexNode, RexUtil}
import org.apache.calcite.sql.`type`.{BasicSqlType, SqlTypeName}
import org.apache.calcite.sql.`type`.SqlTypeName._
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
-import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse}
+import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
import java.math.BigDecimal
@@ -196,18 +196,8 @@ class FlinkRexUtilTest {
)
// the number of RexCall in the CNF result exceeds 95 * 2, so returns the original expression
- val newPredicate1 = FlinkRexUtil.toCnf(rexBuilder, -1, predicate)
- assertEquals(predicate.toString, newPredicate1.toString)
-
- val newPredicate2 = FlinkRexUtil.toCnf(rexBuilder, 200, predicate)
- assertEquals(predicate.toString, newPredicate2.toString)
-
- val newPredicate3 = FlinkRexUtil.toCnf(rexBuilder, 2103039, predicate)
- assertEquals(RexUtil.toCnf(rexBuilder, predicate).toString, newPredicate3.toString)
-
- val newPredicate4 = FlinkRexUtil.toCnf(rexBuilder, Int.MaxValue, predicate)
- assertFalse(predicate.equals(newPredicate4))
- assertEquals(RexUtil.toCnf(rexBuilder, predicate).toString, newPredicate4.toString)
+ val newPredicate = FlinkRexUtil.toCnf(rexBuilder, predicate)
+ assertEquals(predicate.toString, newPredicate.toString)
}
@Test
@@ -239,12 +229,9 @@ class FlinkRexUtilTest {
rexBuilder.makeCall(EQUALS, c, rexBuilder.makeLiteral("3")))
)
- val newPredicate1 = FlinkRexUtil.toCnf(rexBuilder, -1, predicate)
- assertEquals(expected.toString, newPredicate1.toString)
+ val newPredicate = FlinkRexUtil.toCnf(rexBuilder, predicate)
+ assertEquals(expected.toString, newPredicate.toString)
assertEquals(expected.toString, RexUtil.toCnf(rexBuilder, predicate).toString)
-
- val newPredicate2 = FlinkRexUtil.toCnf(rexBuilder, 0, predicate)
- assertEquals(predicate.toString, newPredicate2.toString)
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
index 991d7fe3700..4af1f7be9a6 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
@@ -120,7 +120,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val expected: Array[Expression] = Array(firstExp, secondExp)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(expr, -1, allFieldNames, builder, functionCatalog)
+ extractConjunctiveConditions(expr, allFieldNames, builder, functionCatalog)
assertExpressionArrayEquals(expected, convertedExpressions)
assertThat(unconvertedRexNodes).isEmpty()
@@ -138,7 +138,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(a, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(a, allFieldNames, relBuilder, functionCatalog)
val expected: Array[Expression] = Array($"amount" >= $"id")
assertExpressionArrayEquals(expected, convertedExpressions)
@@ -158,7 +158,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(a, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(a, allFieldNames, relBuilder, functionCatalog)
val expected: Array[Expression] = Array($"blob" === Array[Byte](97, 98, 99))
assertExpressionArrayEquals(expected, convertedExpressions)
@@ -202,7 +202,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(complexNode, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(complexNode, allFieldNames, relBuilder, functionCatalog)
val expected: Array[Expression] = Array(
$"amount" < 100 || $"price" === 100 || $"price" === 200,
@@ -238,7 +238,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(and, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(and, allFieldNames, relBuilder, functionCatalog)
val expected: Array[Expression] = Array(
$"amount" < 100,
@@ -278,7 +278,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(and, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(and, allFieldNames, relBuilder, functionCatalog)
val expected: Array[Expression] = Array(
// amount < 100
@@ -317,7 +317,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (converted, _) =
- extractConjunctiveConditions(and, -1, fieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(and, fieldNames, relBuilder, functionCatalog)
val datetime = DateTimeTestUtil.localDateTime("2017-09-10 14:23:01")
val date = DateTimeTestUtil.localDate("2017-09-12")
@@ -388,7 +388,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val complexExpr = rexBuilder.makeCall(SqlStdOperatorTable.AND, condition)
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(complexExpr, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(complexExpr, allFieldNames, relBuilder, functionCatalog)
val expected: Array[Expression] = Array(
$"amount" < $"id",
@@ -442,7 +442,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(complexExpr, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(complexExpr, allFieldNames, relBuilder, functionCatalog)
{
val expected: Array[Expression] = Array(
@@ -486,7 +486,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(conditionExpr, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(conditionExpr, allFieldNames, relBuilder, functionCatalog)
assertThat(convertedExpressions).hasSize(3)
assertThat(convertedExpressions(0).toString).isEqualTo("greaterThan(cast(amount, BIGINT), 100)")
@@ -542,7 +542,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(and, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(and, allFieldNames, relBuilder, functionCatalog)
assertThat(convertedExpressions).isEmpty()
assertThat(unconvertedRexNodes).hasSize(3)
@@ -564,7 +564,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(condition, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(condition, allFieldNames, relBuilder, functionCatalog)
assertThat(convertedExpressions).hasSize(1)
assertThat(convertedExpressions(0).toString).isEqualTo("greaterThan(MyUdf(amount), 100)")
@@ -589,7 +589,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate1, nonPartitionPredicate1) =
RexNodeExtractor.extractPartitionPredicates(
c3,
- -1,
allFieldNames.asScala.toArray,
rexBuilder,
Array("amount", "name")
@@ -600,7 +599,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate2, nonPartitionPredicate2) =
RexNodeExtractor.extractPartitionPredicates(
c3,
- -1,
allFieldNames.asScala.toArray,
rexBuilder,
Array("amount")
@@ -611,7 +609,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate3, nonPartitionPredicate3) =
RexNodeExtractor.extractPartitionPredicates(
c3,
- -1,
allFieldNames.asScala.toArray,
rexBuilder,
Array("id")
@@ -624,7 +621,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate4, nonPartitionPredicate4) =
RexNodeExtractor.extractPartitionPredicates(
c4,
- -1,
allFieldNames.asScala.toArray,
rexBuilder,
Array("amount", "name")
@@ -635,7 +631,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate5, nonPartitionPredicate5) =
RexNodeExtractor.extractPartitionPredicates(
c4,
- -1,
allFieldNames.asScala.toArray,
rexBuilder,
Array("amount")
@@ -663,7 +658,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate1, nonPartitionPredicate1) =
RexNodeExtractor.extractPartitionPredicates(
c3,
- -1,
Array("date", "amount", "id"),
rexBuilder,
Array("date")
@@ -674,7 +668,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate2, nonPartitionPredicate2) =
RexNodeExtractor.extractPartitionPredicates(
c3,
- -1,
Array("date", "amount", "id"),
rexBuilder,
Array("date", "amount")
@@ -685,7 +678,6 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val (partitionPredicate3, nonPartitionPredicate3) =
RexNodeExtractor.extractPartitionPredicates(
c3,
- -1,
Array("date", "amount", "id"),
rexBuilder,
Array("id", "amount")
@@ -717,7 +709,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (converted, _) =
- extractConjunctiveConditions(and, -1, fieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(and, fieldNames, relBuilder, functionCatalog)
val datetime = DateTimeTestUtil.localDateTime("2017-09-10 14:23:01.123456")
val instant =
@@ -746,7 +738,7 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val conditionExpr = rexBuilder.makeCall(op, t0)
val relBuilder: RexBuilder = new FlinkRexBuilder(typeFactory)
val (convertedExpressions, unconvertedRexNodes) =
- extractConjunctiveConditions(conditionExpr, -1, allFieldNames, relBuilder, functionCatalog)
+ extractConjunctiveConditions(conditionExpr, allFieldNames, relBuilder, functionCatalog)
assertThat(convertedExpressions).hasSize(1)
assertThat(convertedExpressions.head.toString).isEqualTo(expr)
@@ -767,13 +759,11 @@ class RexNodeExtractorTest extends RexNodeTestBase {
private def extractConjunctiveConditions(
expr: RexNode,
- maxCnfNodeCount: Int,
inputFieldNames: JList[String],
rexBuilder: RexBuilder,
catalog: FunctionCatalog): (Array[Expression], Array[RexNode]) = {
RexNodeExtractor.extractConjunctiveConditions(
expr,
- maxCnfNodeCount,
inputFieldNames,
rexBuilder,
catalog,
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
index b2e13cb81f7..32c3b505d10 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
@@ -32,7 +32,6 @@ import org.apache.flink.table.data.{DecimalDataUtils, TimestampData}
import org.apache.flink.table.data.util.DataFormatConverters.LocalDateConverter
import org.apache.flink.table.planner.expressions.utils.{RichFunc1, RichFunc2, RichFunc3, SplitUDF}
import org.apache.flink.table.planner.factories.TestValuesTableFactory
-import org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalSortRule
import org.apache.flink.table.planner.runtime.utils.{BatchTableEnvUtil, BatchTestBase, TestData, UserDefinedFunctionTestUtils}
import org.apache.flink.table.planner.runtime.utils.BatchTableEnvUtil.parseFieldNames
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
@@ -1578,22 +1577,19 @@ class CalcITCase extends BatchTestBase {
nullablesOfNullData3
)
tableConfig.set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, Int.box(1))
- tableConfig.set(BatchPhysicalSortRule.TABLE_EXEC_RANGE_SORT_ENABLED, Boolean.box(true))
- assertThatThrownBy(
- () =>
- checkResult(
- "select * from BinaryT order by c",
- nullData3
- .sortBy((x: Row) => x.getField(2).asInstanceOf[String])
- .map(
- r =>
- row(
- r.getField(0),
- r.getField(1),
- r.getField(2).toString.getBytes(StandardCharsets.UTF_8))),
- isSorted = true
- )).isInstanceOf(classOf[UnsupportedOperationException])
+ checkResult(
+ "select * from BinaryT order by c",
+ nullData3
+ .sortBy((x: Row) => x.getField(2).asInstanceOf[String])
+ .map(
+ r =>
+ row(
+ r.getField(0),
+ r.getField(1),
+ r.getField(2).toString.getBytes(StandardCharsets.UTF_8))),
+ isSorted = true
+ )
}
@Test