You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2022/02/25 14:51:33 UTC
[flink] 01/02: [FLINK-26287][table] Add convenience method TableConfig.set
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d2e947361d6e413f28b2cccebeec2e1761bc253b
Author: Timo Walther <tw...@apache.org>
AuthorDate: Tue Feb 22 17:30:45 2022 +0100
[FLINK-26287][table] Add convenience method TableConfig.set
---
.../docs/dev/python/dependency_management.md | 2 +-
docs/content.zh/docs/dev/table/config.md | 24 ++++----
.../docs/dev/table/sql/queries/overview.md | 12 ++--
docs/content.zh/docs/dev/table/tuning.md | 41 ++++++-------
.../docs/dev/python/dependency_management.md | 2 +-
docs/content/docs/dev/table/config.md | 24 ++++----
.../content/docs/dev/table/sql/queries/overview.md | 8 +--
docs/content/docs/dev/table/tuning.md | 69 +++++++++++-----------
flink-python/pyflink/table/table_config.py | 9 +++
.../org/apache/flink/table/api/TableConfig.java | 49 +++++++++++++--
10 files changed, 141 insertions(+), 99 deletions(-)
diff --git a/docs/content.zh/docs/dev/python/dependency_management.md b/docs/content.zh/docs/dev/python/dependency_management.md
index 14ca3cd..c818686 100644
--- a/docs/content.zh/docs/dev/python/dependency_management.md
+++ b/docs/content.zh/docs/dev/python/dependency_management.md
@@ -302,7 +302,7 @@ import org.apache.flink.table.api.TableEnvironment;
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inBatchMode());
-tEnv.getConfig().getConfiguration().set(CoreOptions.DEFAULT_PARALLELISM, 1);
+tEnv.getConfig().set(CoreOptions.DEFAULT_PARALLELISM, 1);
// register the Python UDF
tEnv.executeSql("create temporary system function add_one as 'add_one.add_one' language python");
diff --git a/docs/content.zh/docs/dev/table/config.md b/docs/content.zh/docs/dev/table/config.md
index 2a4ac8f..70939a9 100644
--- a/docs/content.zh/docs/dev/table/config.md
+++ b/docs/content.zh/docs/dev/table/config.md
@@ -49,11 +49,11 @@ Table 和 SQL API 的默认配置能够确保结果准确，同时也提供可
TableEnvironment tEnv = ...
// access flink configuration
-Configuration configuration = tEnv.getConfig().getConfiguration();
+TableConfig configuration = tEnv.getConfig();
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true");
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
-configuration.setString("table.exec.mini-batch.size", "5000");
+configuration.set("table.exec.mini-batch.enabled", "true");
+configuration.set("table.exec.mini-batch.allow-latency", "5 s");
+configuration.set("table.exec.mini-batch.size", "5000");
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -62,11 +62,11 @@ configuration.setString("table.exec.mini-batch.size", "5000");
val tEnv: TableEnvironment = ...
// access flink configuration
-val configuration = tEnv.getConfig().getConfiguration()
+val configuration = tEnv.getConfig()
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true")
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
-configuration.setString("table.exec.mini-batch.size", "5000")
+configuration.set("table.exec.mini-batch.enabled", "true")
+configuration.set("table.exec.mini-batch.allow-latency", "5 s")
+configuration.set("table.exec.mini-batch.size", "5000")
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -75,11 +75,11 @@ configuration.setString("table.exec.mini-batch.size", "5000")
t_env = ...
# access flink configuration
-configuration = t_env.get_config().get_configuration()
+configuration = t_env.get_config()
# set low-level key-value options
-configuration.set_string("table.exec.mini-batch.enabled", "true")
-configuration.set_string("table.exec.mini-batch.allow-latency", "5 s")
-configuration.set_string("table.exec.mini-batch.size", "5000")
+configuration.set("table.exec.mini-batch.enabled", "true")
+configuration.set("table.exec.mini-batch.allow-latency", "5 s")
+configuration.set("table.exec.mini-batch.size", "5000")
```
{{< /tab >}}
{{< tab "SQL CLI" >}}
diff --git a/docs/content.zh/docs/dev/table/sql/queries/overview.md b/docs/content.zh/docs/dev/table/sql/queries/overview.md
index 4d54444..a1074a8 100644
--- a/docs/content.zh/docs/dev/table/sql/queries/overview.md
+++ b/docs/content.zh/docs/dev/table/sql/queries/overview.md
@@ -205,10 +205,10 @@ tableResult2.print();
val env = StreamExecutionEnvironment.getExecutionEnvironment()
val tableEnv = StreamTableEnvironment.create(env, settings)
// enable checkpointing
-tableEnv.getConfig.getConfiguration.set(
- ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE)
-tableEnv.getConfig.getConfiguration.set(
- ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10))
+tableEnv.getConfig
+ .set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE)
+tableEnv.getConfig
+ .set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10))
tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
@@ -232,8 +232,8 @@ tableResult2.print()
env = StreamExecutionEnvironment.get_execution_environment()
table_env = StreamTableEnvironment.create(env, settings)
# enable checkpointing
-table_env.get_config().get_configuration().set_string("execution.checkpointing.mode", "EXACTLY_ONCE")
-table_env.get_config().get_configuration().set_string("execution.checkpointing.interval", "10s")
+table_env.get_config().set("execution.checkpointing.mode", "EXACTLY_ONCE")
+table_env.get_config().set("execution.checkpointing.interval", "10s")
table_env.execute_sql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
diff --git a/docs/content.zh/docs/dev/table/tuning.md b/docs/content.zh/docs/dev/table/tuning.md
index 0031f5b..5d3111a 100644
--- a/docs/content.zh/docs/dev/table/tuning.md
+++ b/docs/content.zh/docs/dev/table/tuning.md
@@ -62,11 +62,11 @@ Window TVF aggregation buffer records in [managed memory]({{< ref "docs/deployme
TableEnvironment tEnv = ...
// access flink configuration
-Configuration configuration = tEnv.getConfig().getConfiguration();
+TableConfig configuration = tEnv.getConfig();
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true"); // enable mini-batch optimization
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s"); // use 5 seconds to buffer input records
-configuration.setString("table.exec.mini-batch.size", "5000"); // the maximum number of records can be buffered by each aggregate operator task
+configuration.set("table.exec.mini-batch.enabled", "true"); // enable mini-batch optimization
+configuration.set("table.exec.mini-batch.allow-latency", "5 s"); // use 5 seconds to buffer input records
+configuration.set("table.exec.mini-batch.size", "5000"); // the maximum number of records can be buffered by each aggregate operator task
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -75,11 +75,11 @@ configuration.setString("table.exec.mini-batch.size", "5000"); // the maximum nu
val tEnv: TableEnvironment = ...
// access flink configuration
-val configuration = tEnv.getConfig().getConfiguration()
+val configuration = tEnv.getConfig()
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true") // enable mini-batch optimization
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s") // use 5 seconds to buffer input records
-configuration.setString("table.exec.mini-batch.size", "5000") // the maximum number of records can be buffered by each aggregate operator task
+configuration.set("table.exec.mini-batch.enabled", "true") // enable mini-batch optimization
+configuration.set("table.exec.mini-batch.allow-latency", "5 s") // use 5 seconds to buffer input records
+configuration.set("table.exec.mini-batch.size", "5000") // the maximum number of records can be buffered by each aggregate operator task
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -138,12 +138,12 @@ configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE"); // e
val tEnv: TableEnvironment = ...
// access flink configuration
-val configuration = tEnv.getConfig().getConfiguration()
+val configuration = tEnv.getConfig()
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true") // local-global aggregation depends on mini-batch is enabled
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
-configuration.setString("table.exec.mini-batch.size", "5000")
-configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE") // enable two-phase, i.e. local-global aggregation
+configuration.set("table.exec.mini-batch.enabled", "true") // local-global aggregation depends on mini-batch is enabled
+configuration.set("table.exec.mini-batch.allow-latency", "5 s")
+configuration.set("table.exec.mini-batch.size", "5000")
+configuration.set("table.optimizer.agg-phase-strategy", "TWO_PHASE") // enable two-phase, i.e. local-global aggregation
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -208,9 +208,8 @@ GROUP BY day
// instantiate table environment
TableEnvironment tEnv = ...
-tEnv.getConfig() // access high-level configuration
- .getConfiguration() // set low-level key-value options
- .setString("table.optimizer.distinct-agg.split.enabled", "true"); // enable distinct agg split
+tEnv.getConfig()
+ .set("table.optimizer.distinct-agg.split.enabled", "true"); // enable distinct agg split
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -218,9 +217,8 @@ tEnv.getConfig() // access high-level configuration
// instantiate table environment
val tEnv: TableEnvironment = ...
-tEnv.getConfig // access high-level configuration
- .getConfiguration // set low-level key-value options
- .setString("table.optimizer.distinct-agg.split.enabled", "true") // enable distinct agg split
+tEnv.getConfig
+ .set("table.optimizer.distinct-agg.split.enabled", "true") // enable distinct agg split
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -228,9 +226,8 @@ tEnv.getConfig // access high-level configuration
# instantiate table environment
t_env = ...
-t_env.get_config() # access high-level configuration
- .get_configuration() # set low-level key-value options
- .set_string("table.optimizer.distinct-agg.split.enabled", "true"); # enable distinct agg split
+t_env.get_config()
+ .set("table.optimizer.distinct-agg.split.enabled", "true"); # enable distinct agg split
```
{{< /tab >}}
{{< /tabs >}}
diff --git a/docs/content/docs/dev/python/dependency_management.md b/docs/content/docs/dev/python/dependency_management.md
index 7ed3c1f..e27ec80 100644
--- a/docs/content/docs/dev/python/dependency_management.md
+++ b/docs/content/docs/dev/python/dependency_management.md
@@ -302,7 +302,7 @@ import org.apache.flink.table.api.TableEnvironment;
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inBatchMode());
-tEnv.getConfig().getConfiguration().set(CoreOptions.DEFAULT_PARALLELISM, 1);
+tEnv.getConfig().set(CoreOptions.DEFAULT_PARALLELISM, 1);
// register the Python UDF
tEnv.executeSql("create temporary system function add_one as 'add_one.add_one' language python");
diff --git a/docs/content/docs/dev/table/config.md b/docs/content/docs/dev/table/config.md
index e6ebdd3..9cc928d 100644
--- a/docs/content/docs/dev/table/config.md
+++ b/docs/content/docs/dev/table/config.md
@@ -54,11 +54,11 @@ table environment.
TableEnvironment tEnv = ...
// access flink configuration
-Configuration configuration = tEnv.getConfig().getConfiguration();
+TableConfig configuration = tEnv.getConfig();
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true");
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
-configuration.setString("table.exec.mini-batch.size", "5000");
+configuration.set("table.exec.mini-batch.enabled", "true");
+configuration.set("table.exec.mini-batch.allow-latency", "5 s");
+configuration.set("table.exec.mini-batch.size", "5000");
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -67,11 +67,11 @@ configuration.setString("table.exec.mini-batch.size", "5000");
val tEnv: TableEnvironment = ...
// access flink configuration
-val configuration = tEnv.getConfig().getConfiguration()
+val configuration = tEnv.getConfig()
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true")
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
-configuration.setString("table.exec.mini-batch.size", "5000")
+configuration.set("table.exec.mini-batch.enabled", "true")
+configuration.set("table.exec.mini-batch.allow-latency", "5 s")
+configuration.set("table.exec.mini-batch.size", "5000")
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -80,11 +80,11 @@ configuration.setString("table.exec.mini-batch.size", "5000")
t_env = ...
# access flink configuration
-configuration = t_env.get_config().get_configuration()
+configuration = t_env.get_config()
# set low-level key-value options
-configuration.set_string("table.exec.mini-batch.enabled", "true")
-configuration.set_string("table.exec.mini-batch.allow-latency", "5 s")
-configuration.set_string("table.exec.mini-batch.size", "5000")
+configuration.set("table.exec.mini-batch.enabled", "true")
+configuration.set("table.exec.mini-batch.allow-latency", "5 s")
+configuration.set("table.exec.mini-batch.size", "5000")
```
{{< /tab >}}
{{< tab "SQL CLI" >}}
diff --git a/docs/content/docs/dev/table/sql/queries/overview.md b/docs/content/docs/dev/table/sql/queries/overview.md
index 70cada6..d59fe90 100644
--- a/docs/content/docs/dev/table/sql/queries/overview.md
+++ b/docs/content/docs/dev/table/sql/queries/overview.md
@@ -205,9 +205,9 @@ tableResult2.print();
val env = StreamExecutionEnvironment.getExecutionEnvironment()
val tableEnv = StreamTableEnvironment.create(env, settings)
// enable checkpointing
-tableEnv.getConfig.getConfiguration.set(
+tableEnv.getConfig.set(
ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE)
-tableEnv.getConfig.getConfiguration.set(
+tableEnv.getConfig.set(
ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10))
tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
@@ -232,8 +232,8 @@ tableResult2.print()
env = StreamExecutionEnvironment.get_execution_environment()
table_env = StreamTableEnvironment.create(env, settings)
# enable checkpointing
-table_env.get_config().get_configuration().set_string("execution.checkpointing.mode", "EXACTLY_ONCE")
-table_env.get_config().get_configuration().set_string("execution.checkpointing.interval", "10s")
+table_env.get_config().set("execution.checkpointing.mode", "EXACTLY_ONCE")
+table_env.get_config().set("execution.checkpointing.interval", "10s")
table_env.execute_sql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
diff --git a/docs/content/docs/dev/table/tuning.md b/docs/content/docs/dev/table/tuning.md
index b93a868..804e75f 100644
--- a/docs/content/docs/dev/table/tuning.md
+++ b/docs/content/docs/dev/table/tuning.md
@@ -62,11 +62,11 @@ The following examples show how to enable these options.
TableEnvironment tEnv = ...
// access flink configuration
-Configuration configuration = tEnv.getConfig().getConfiguration();
+TableConfig configuration = tEnv.getConfig();
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true"); // enable mini-batch optimization
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s"); // use 5 seconds to buffer input records
-configuration.setString("table.exec.mini-batch.size", "5000"); // the maximum number of records can be buffered by each aggregate operator task
+configuration.set("table.exec.mini-batch.enabled", "true"); // enable mini-batch optimization
+configuration.set("table.exec.mini-batch.allow-latency", "5 s"); // use 5 seconds to buffer input records
+configuration.set("table.exec.mini-batch.size", "5000"); // the maximum number of records can be buffered by each aggregate operator task
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -75,11 +75,11 @@ configuration.setString("table.exec.mini-batch.size", "5000"); // the maximum nu
val tEnv: TableEnvironment = ...
// access flink configuration
-val configuration = tEnv.getConfig().getConfiguration()
+val configuration = tEnv.getConfig()
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true") // enable mini-batch optimization
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s") // use 5 seconds to buffer input records
-configuration.setString("table.exec.mini-batch.size", "5000") // the maximum number of records can be buffered by each aggregate operator task
+configuration.set("table.exec.mini-batch.enabled", "true") // enable mini-batch optimization
+configuration.set("table.exec.mini-batch.allow-latency", "5 s") // use 5 seconds to buffer input records
+configuration.set("table.exec.mini-batch.size", "5000") // the maximum number of records can be buffered by each aggregate operator task
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -88,11 +88,11 @@ configuration.setString("table.exec.mini-batch.size", "5000") // the maximum num
t_env = ...
# access flink configuration
-configuration = t_env.get_config().get_configuration();
+configuration = t_env.get_config();
# set low-level key-value options
-configuration.set_string("table.exec.mini-batch.enabled", "true"); # enable mini-batch optimization
-configuration.set_string("table.exec.mini-batch.allow-latency", "5 s"); # use 5 seconds to buffer input records
-configuration.set_string("table.exec.mini-batch.size", "5000"); # the maximum number of records can be buffered by each aggregate operator task
+configuration.set("table.exec.mini-batch.enabled", "true"); # enable mini-batch optimization
+configuration.set("table.exec.mini-batch.allow-latency", "5 s"); # use 5 seconds to buffer input records
+configuration.set("table.exec.mini-batch.size", "5000"); # the maximum number of records can be buffered by each aggregate operator task
```
{{< /tab >}}
{{< /tabs >}}
@@ -125,12 +125,12 @@ The following examples show how to enable the local-global aggregation.
TableEnvironment tEnv = ...
// access flink configuration
-Configuration configuration = tEnv.getConfig().getConfiguration();
+TableConfig configuration = tEnv.getConfig();
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true"); // local-global aggregation depends on mini-batch is enabled
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
-configuration.setString("table.exec.mini-batch.size", "5000");
-configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE"); // enable two-phase, i.e. local-global aggregation
+configuration.set("table.exec.mini-batch.enabled", "true"); // local-global aggregation depends on mini-batch is enabled
+configuration.set("table.exec.mini-batch.allow-latency", "5 s");
+configuration.set("table.exec.mini-batch.size", "5000");
+configuration.set("table.optimizer.agg-phase-strategy", "TWO_PHASE"); // enable two-phase, i.e. local-global aggregation
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -139,12 +139,12 @@ configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE"); // e
val tEnv: TableEnvironment = ...
// access flink configuration
-val configuration = tEnv.getConfig().getConfiguration()
+val configuration = tEnv.getConfig()
// set low-level key-value options
-configuration.setString("table.exec.mini-batch.enabled", "true") // local-global aggregation depends on mini-batch is enabled
-configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
-configuration.setString("table.exec.mini-batch.size", "5000")
-configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE") // enable two-phase, i.e. local-global aggregation
+configuration.set("table.exec.mini-batch.enabled", "true") // local-global aggregation depends on mini-batch is enabled
+configuration.set("table.exec.mini-batch.allow-latency", "5 s")
+configuration.set("table.exec.mini-batch.size", "5000")
+configuration.set("table.optimizer.agg-phase-strategy", "TWO_PHASE") // enable two-phase, i.e. local-global aggregation
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -153,12 +153,12 @@ configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE") // en
t_env = ...
# access flink configuration
-configuration = t_env.get_config().get_configuration();
+configuration = t_env.get_config();
# set low-level key-value options
-configuration.set_string("table.exec.mini-batch.enabled", "true"); # local-global aggregation depends on mini-batch is enabled
-configuration.set_string("table.exec.mini-batch.allow-latency", "5 s");
-configuration.set_string("table.exec.mini-batch.size", "5000");
-configuration.set_string("table.optimizer.agg-phase-strategy", "TWO_PHASE"); # enable two-phase, i.e. local-global aggregation
+configuration.set("table.exec.mini-batch.enabled", "true"); # local-global aggregation depends on mini-batch is enabled
+configuration.set("table.exec.mini-batch.allow-latency", "5 s");
+configuration.set("table.exec.mini-batch.size", "5000");
+configuration.set("table.optimizer.agg-phase-strategy", "TWO_PHASE"); # enable two-phase, i.e. local-global aggregation
```
{{< /tab >}}
{{< /tabs >}}
@@ -212,9 +212,8 @@ The following examples show how to enable the split distinct aggregation optimiz
// instantiate table environment
TableEnvironment tEnv = ...
-tEnv.getConfig() // access high-level configuration
- .getConfiguration() // set low-level key-value options
- .setString("table.optimizer.distinct-agg.split.enabled", "true"); // enable distinct agg split
+tEnv.getConfig()
+ .set("table.optimizer.distinct-agg.split.enabled", "true"); // enable distinct agg split
```
{{< /tab >}}
{{< tab "Scala" >}}
@@ -222,9 +221,8 @@ tEnv.getConfig() // access high-level configuration
// instantiate table environment
val tEnv: TableEnvironment = ...
-tEnv.getConfig // access high-level configuration
- .getConfiguration // set low-level key-value options
- .setString("table.optimizer.distinct-agg.split.enabled", "true") // enable distinct agg split
+tEnv.getConfig
+ .set("table.optimizer.distinct-agg.split.enabled", "true") // enable distinct agg split
```
{{< /tab >}}
{{< tab "Python" >}}
@@ -232,9 +230,8 @@ tEnv.getConfig // access high-level configuration
# instantiate table environment
t_env = ...
-t_env.get_config() # access high-level configuration
- .get_configuration() # set low-level key-value options
- .set_string("table.optimizer.distinct-agg.split.enabled", "true"); # enable distinct agg split
+t_env.get_config()
+ .set("table.optimizer.distinct-agg.split.enabled", "true"); # enable distinct agg split
```
{{< /tab >}}
{{< /tabs >}}
diff --git a/flink-python/pyflink/table/table_config.py b/flink-python/pyflink/table/table_config.py
index f4ebe6c..eacd635 100644
--- a/flink-python/pyflink/table/table_config.py
+++ b/flink-python/pyflink/table/table_config.py
@@ -51,6 +51,15 @@ class TableConfig(object):
else:
self._j_table_config = j_table_config
+ def set(self, key: str, value: str) -> 'TableConfig':
+ """
+ Sets a string-based value for the given string-based key.
+
+ The value will be parsed by the framework on access.
+ """
+ self._j_table_config.set(key, value)
+ return self
+
def get_local_timezone(self) -> str:
"""
Returns the local timezone id for timestamp with local time zone, either an abbreviation
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
index ba0a2ac..2565bb4 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
@@ -21,8 +21,10 @@ package org.apache.flink.table.api;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.configuration.WritableConfig;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.api.config.TableConfigOptions;
@@ -61,11 +63,12 @@ import static java.time.ZoneId.SHORT_IDS;
* <p>Note: Because options are read at different point in time when performing operations, it is
* recommended to set configuration options early after instantiating a table environment.
*
+ * @see TableConfigOptions
* @see ExecutionConfigOptions
* @see OptimizerConfigOptions
*/
@PublicEvolving
-public class TableConfig {
+public class TableConfig implements WritableConfig {
/** Defines if all fields need to be checked for NULL first. */
private Boolean nullCheck = true;
@@ -81,6 +84,43 @@ public class TableConfig {
/** A configuration object to hold all key/value configuration. */
private final Configuration configuration = new Configuration();
+ /**
+ * Sets a value for the given {@link ConfigOption}.
+ *
+ * <p>This method should be preferred over {@link #set(String, String)} as it is type-safe,
+ * avoids unnecessary parsing of the value, and provides inline documentation.
+ *
+ * <p>Note: Scala users might need to convert the value into a boxed type. E.g. by using {@code
+ * Int.box(1)} or {@code Boolean.box(false)}.
+ *
+ * @see TableConfigOptions
+ * @see ExecutionConfigOptions
+ * @see OptimizerConfigOptions
+ */
+ @Override
+ public <T> TableConfig set(ConfigOption<T> option, T value) {
+ configuration.set(option, value);
+ return this;
+ }
+
+ /**
+ * Sets a string-based value for the given string-based key.
+ *
+ * <p>The value will be parsed by the framework on access.
+ *
+ * <p>This method exists for convenience when configuring a session with string-based
+ * properties. Use {@link #set(ConfigOption, Object)} for more type-safety and inline
+ * documentation.
+ *
+ * @see TableConfigOptions
+ * @see ExecutionConfigOptions
+ * @see OptimizerConfigOptions
+ */
+ public TableConfig set(String key, String value) {
+ configuration.setString(key, value);
+ return this;
+ }
+
/** Gives direct access to the underlying key-value map for advanced configuration. */
public Configuration getConfiguration() {
return configuration;
@@ -105,8 +145,7 @@ public class TableConfig {
/** Sets the current SQL dialect to parse a SQL query. Flink's SQL behavior by default. */
public void setSqlDialect(SqlDialect sqlDialect) {
- getConfiguration()
- .setString(TableConfigOptions.TABLE_SQL_DIALECT, sqlDialect.name().toLowerCase());
+ set(TableConfigOptions.TABLE_SQL_DIALECT, sqlDialect.name().toLowerCase());
}
/**
@@ -353,7 +392,7 @@ public class TableConfig {
* <pre>{@code
* Map<String, String> params = ...
* TableConfig tableConfig = tEnv.getConfig;
- * tableConfig.getConfiguration().set(PipelineOptions.GLOBAL_JOB_PARAMETERS, params);
+ * config.set(PipelineOptions.GLOBAL_JOB_PARAMETERS, params);
* }</pre>
*/
@Experimental
@@ -364,7 +403,7 @@ public class TableConfig {
.map(HashMap::new)
.orElseGet(HashMap::new);
params.put(key, value);
- getConfiguration().set(PipelineOptions.GLOBAL_JOB_PARAMETERS, params);
+ set(PipelineOptions.GLOBAL_JOB_PARAMETERS, params);
}
public static TableConfig getDefault() {