This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 2385cc05e31 [FLINK-34918][table] Support `ALTER CATALOG COMMENT` syntax
2385cc05e31 is described below
commit 2385cc05e311797d209e85ce9b9e668a7dd3c51c
Author: Yubin Li <[email protected]>
AuthorDate: Thu Jun 13 15:55:20 2024 +0800
[FLINK-34918][table] Support `ALTER CATALOG COMMENT` syntax
---
docs/content.zh/docs/dev/table/sql/alter.md | 11 ++++
docs/content/docs/dev/table/sql/alter.md | 11 ++++
.../src/test/resources/sql/catalog_database.q | 30 +++++++++
.../src/test/resources/sql/catalog_database.q | 44 +++++++++++++
.../src/main/codegen/data/Parser.tdd | 1 +
.../src/main/codegen/includes/parserImpls.ftl | 9 +++
.../sql/parser/ddl/SqlAlterCatalogComment.java | 57 ++++++++++++++++
.../flink/sql/parser/FlinkSqlParserImplTest.java | 1 +
.../apache/flink/table/catalog/CatalogManager.java | 19 +++---
...tion.java => AlterCatalogCommentOperation.java} | 30 ++++-----
.../ddl/AlterCatalogOptionsOperation.java | 6 +-
.../operations/ddl/AlterCatalogResetOperation.java | 7 +-
.../flink/table/catalog/CatalogManagerTest.java | 29 +++++++++
.../apache/flink/table/catalog/CatalogChange.java | 75 ++++++++++++++++++++++
.../flink/table/catalog/CatalogDescriptor.java | 5 ++
.../SqlAlterCatalogCommentConverter.java | 40 ++++++++++++
.../operations/converters/SqlNodeConverters.java | 1 +
.../operations/SqlDdlToOperationConverterTest.java | 15 +++++
18 files changed, 361 insertions(+), 30 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/sql/alter.md
b/docs/content.zh/docs/dev/table/sql/alter.md
index 39de6985c9d..993613379e4 100644
--- a/docs/content.zh/docs/dev/table/sql/alter.md
+++ b/docs/content.zh/docs/dev/table/sql/alter.md
@@ -547,6 +547,7 @@ Language tag 用于指定 Flink runtime 如何执行这个函数。目前，只
ALTER CATALOG catalog_name
SET (key1=val1, ...)
| RESET (key1, ...)
+ | COMMENT 'comment'
```
### SET
@@ -571,4 +572,14 @@ ALTER CATALOG cat2 SET ('default-database'='db');
ALTER CATALOG cat2 RESET ('default-database');
```
+### COMMENT
+
+为指定的 catalog 设置注释。若注释已经存在，则使用新值覆盖旧值。
+
+`COMMENT` 语句示例如下。
+
+```sql
+ALTER CATALOG cat2 COMMENT 'comment for catalog ''cat2''';
+```
+
{{< top >}}
diff --git a/docs/content/docs/dev/table/sql/alter.md
b/docs/content/docs/dev/table/sql/alter.md
index e842d137ce0..b5f199ced5b 100644
--- a/docs/content/docs/dev/table/sql/alter.md
+++ b/docs/content/docs/dev/table/sql/alter.md
@@ -549,6 +549,7 @@ Language tag to instruct flink runtime how to execute the
function. Currently on
ALTER CATALOG catalog_name
SET (key1=val1, ...)
| RESET (key1, ...)
+ | COMMENT 'comment'
```
### SET
@@ -573,4 +574,14 @@ The following examples illustrate the usage of the `RESET`
statements.
ALTER CATALOG cat2 RESET ('default-database');
```
+### COMMENT
+
+Set comment in the specified catalog. If the comment is already set in the
catalog, override the old value with the new one.
+
+The following examples illustrate the usage of the `COMMENT` statements.
+
+```sql
+ALTER CATALOG cat2 COMMENT 'comment for catalog ''cat2''';
+```
+
{{< top >}}
diff --git
a/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
b/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
index a2ae9a2fe1a..93c219f4e52 100644
--- a/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
+++ b/flink-table/flink-sql-client/src/test/resources/sql/catalog_database.q
@@ -246,6 +246,36 @@ alter catalog cat2 reset ();
org.apache.flink.table.api.ValidationException: ALTER CATALOG RESET does not
support empty key
!error
+alter catalog cat2 comment 'comment for catalog ''cat2''';
+[INFO] Execute statement succeeded.
+!info
+
+desc catalog extended cat2;
++-----------+----------------------------+
+| info name | info value |
++-----------+----------------------------+
+| name | cat2 |
+| type | generic_in_memory |
+| comment | comment for catalog 'cat2' |
++-----------+----------------------------+
+3 rows in set
+!ok
+
+alter catalog cat2 comment '';
+[INFO] Execute statement succeeded.
+!info
+
+desc catalog extended cat2;
++-----------+-------------------+
+| info name | info value |
++-----------+-------------------+
+| name | cat2 |
+| type | generic_in_memory |
+| comment | |
++-----------+-------------------+
+3 rows in set
+!ok
+
# ==========================================================================
# test database
# ==========================================================================
diff --git
a/flink-table/flink-sql-gateway/src/test/resources/sql/catalog_database.q
b/flink-table/flink-sql-gateway/src/test/resources/sql/catalog_database.q
index 687330e4a50..644c42774e9 100644
--- a/flink-table/flink-sql-gateway/src/test/resources/sql/catalog_database.q
+++ b/flink-table/flink-sql-gateway/src/test/resources/sql/catalog_database.q
@@ -290,6 +290,50 @@ alter catalog cat2 reset ();
org.apache.flink.table.api.ValidationException: ALTER CATALOG RESET does not
support empty key
!error
+alter catalog cat2 comment 'comment for catalog ''cat2''';
+!output
++--------+
+| result |
++--------+
+| OK |
++--------+
+1 row in set
+!ok
+
+desc catalog extended cat2;
+!output
++-----------+----------------------------+
+| info name | info value |
++-----------+----------------------------+
+| name | cat2 |
+| type | generic_in_memory |
+| comment | comment for catalog 'cat2' |
++-----------+----------------------------+
+3 rows in set
+!ok
+
+alter catalog cat2 comment '';
+!output
++--------+
+| result |
++--------+
+| OK |
++--------+
+1 row in set
+!ok
+
+desc catalog extended cat2;
+!output
++-----------+-------------------+
+| info name | info value |
++-----------+-------------------+
+| name | cat2 |
+| type | generic_in_memory |
+| comment | |
++-----------+-------------------+
+3 rows in set
+!ok
+
# ==========================================================================
# test database
# ==========================================================================
diff --git a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd
b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd
index 1bb300efd21..722d7fdf086 100644
--- a/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd
+++ b/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd
@@ -36,6 +36,7 @@
"org.apache.flink.sql.parser.ddl.SqlAlterCatalog"
"org.apache.flink.sql.parser.ddl.SqlAlterCatalogOptions"
"org.apache.flink.sql.parser.ddl.SqlAlterCatalogReset"
+ "org.apache.flink.sql.parser.ddl.SqlAlterCatalogComment"
"org.apache.flink.sql.parser.ddl.SqlAlterDatabase"
"org.apache.flink.sql.parser.ddl.SqlAlterFunction"
"org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTable"
diff --git
a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
index b4ea9664c4d..2c7220a34b1 100644
--- a/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
+++ b/flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl
@@ -166,6 +166,7 @@ SqlAlterCatalog SqlAlterCatalog() :
SqlParserPos startPos;
SqlIdentifier catalogName;
SqlNodeList propertyList = SqlNodeList.EMPTY;
+ SqlNode comment = null;
}
{
<ALTER> <CATALOG> { startPos = getPos(); }
@@ -186,6 +187,14 @@ SqlAlterCatalog SqlAlterCatalog() :
catalogName,
propertyList);
}
+ |
+ <COMMENT>
+ comment = StringLiteral()
+ {
+ return new SqlAlterCatalogComment(startPos.plus(getPos()),
+ catalogName,
+ comment);
+ }
)
}
diff --git
a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterCatalogComment.java
b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterCatalogComment.java
new file mode 100644
index 00000000000..724e1241576
--- /dev/null
+++
b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterCatalogComment.java
@@ -0,0 +1,57 @@
+/*
+ * 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.sql.parser.ddl;
+
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.ImmutableNullableList;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/** ALTER CATALOG catalog_name COMMENT 'comment'. */
+public class SqlAlterCatalogComment extends SqlAlterCatalog {
+
+ private final SqlNode comment;
+
+ public SqlAlterCatalogComment(
+ SqlParserPos position, SqlIdentifier catalogName, SqlNode comment)
{
+ super(position, catalogName);
+ this.comment = requireNonNull(comment, "comment cannot be null");
+ }
+
+ @Override
+ public List<SqlNode> getOperandList() {
+ return ImmutableNullableList.of(catalogName, comment);
+ }
+
+ public SqlNode getComment() {
+ return comment;
+ }
+
+ @Override
+ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+ super.unparse(writer, leftPrec, rightPrec);
+ writer.keyword("COMMENT");
+ comment.unparse(writer, leftPrec, rightPrec);
+ }
+}
diff --git
a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
index 8f75d9f0ad6..08e423899a2 100644
---
a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
+++
b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
@@ -73,6 +73,7 @@ class FlinkSqlParserImplTest extends SqlParserTest {
sql("alter catalog a set ('k1'='v1', 'k2'='v2')")
.ok("ALTER CATALOG `A` SET (\n" + " 'k1' = 'v1',\n" + " 'k2'
= 'v2'\n" + ")");
sql("alter catalog a reset ('k1')").ok("ALTER CATALOG `A` RESET (\n" +
" 'k1'\n" + ")");
+ sql("alter catalog a comment 'comment1'").ok("ALTER CATALOG `A`
COMMENT 'comment1'");
}
@Test
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
index f97a625a476..5491df3f79e 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
@@ -21,7 +21,6 @@ package org.apache.flink.table.catalog;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.CatalogNotExistException;
import org.apache.flink.table.api.EnvironmentSettings;
@@ -334,33 +333,31 @@ public final class CatalogManager implements
CatalogRegistry, AutoCloseable {
* Alters a catalog under the given name. The catalog name must be unique.
*
* @param catalogName the given catalog name under which to alter the
given catalog
- * @param catalogUpdater catalog configuration updater to alter catalog
+ * @param catalogChange catalog change to update the underlying catalog
descriptor
* @throws CatalogException If the catalog neither exists in the catalog
store nor in the
* initialized catalogs, or if an error occurs while creating the
catalog or storing the
* {@link CatalogDescriptor}
*/
- public void alterCatalog(String catalogName, Consumer<Configuration>
catalogUpdater)
+ public void alterCatalog(String catalogName, CatalogChange catalogChange)
throws CatalogException {
checkArgument(
!StringUtils.isNullOrWhitespaceOnly(catalogName),
"Catalog name cannot be null or empty.");
- checkNotNull(catalogUpdater, "Catalog configuration updater cannot be
null.");
+ checkNotNull(catalogChange, "Catalog change cannot be null.");
CatalogStore catalogStore = catalogStoreHolder.catalogStore();
- Optional<CatalogDescriptor> oldCatalogDescriptor =
getCatalogDescriptor(catalogName);
+ Optional<CatalogDescriptor> oldDescriptorOpt =
getCatalogDescriptor(catalogName);
- if (catalogStore.contains(catalogName) &&
oldCatalogDescriptor.isPresent()) {
- Configuration conf = oldCatalogDescriptor.get().getConfiguration();
- catalogUpdater.accept(conf);
- CatalogDescriptor newCatalogDescriptor =
CatalogDescriptor.of(catalogName, conf);
- Catalog newCatalog = initCatalog(catalogName,
newCatalogDescriptor);
+ if (catalogStore.contains(catalogName) &&
oldDescriptorOpt.isPresent()) {
+ CatalogDescriptor newDescriptor =
catalogChange.applyChange(oldDescriptorOpt.get());
+ Catalog newCatalog = initCatalog(catalogName, newDescriptor);
catalogStore.removeCatalog(catalogName, false);
if (catalogs.containsKey(catalogName)) {
catalogs.get(catalogName).close();
}
newCatalog.open();
catalogs.put(catalogName, newCatalog);
- catalogStoreHolder.catalogStore().storeCatalog(catalogName,
newCatalogDescriptor);
+ catalogStoreHolder.catalogStore().storeCatalog(catalogName,
newDescriptor);
} else {
throw new CatalogException(
String.format("Catalog %s does not exist in the catalog
store.", catalogName));
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogResetOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogCommentOperation.java
similarity index 69%
copy from
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogResetOperation.java
copy to
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogCommentOperation.java
index b68387def74..08c6cea11e5 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogResetOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogCommentOperation.java
@@ -22,48 +22,44 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.internal.TableResultImpl;
import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.CatalogChange;
import org.apache.flink.table.catalog.exceptions.CatalogException;
-
-import java.util.Collections;
-import java.util.Set;
-import java.util.stream.Collectors;
+import org.apache.flink.table.utils.EncodingUtils;
import static org.apache.flink.util.Preconditions.checkNotNull;
-/** Operation to describe an ALTER CATALOG RESET statement. */
+/** Operation to describe a ALTER CATALOG COMMENT statement. */
@Internal
-public class AlterCatalogResetOperation implements AlterOperation {
+public class AlterCatalogCommentOperation implements AlterOperation {
+
private final String catalogName;
- private final Set<String> resetKeys;
+ private final String comment;
- public AlterCatalogResetOperation(String catalogName, Set<String>
resetKeys) {
+ public AlterCatalogCommentOperation(String catalogName, String comment) {
this.catalogName = checkNotNull(catalogName);
- this.resetKeys = Collections.unmodifiableSet(checkNotNull(resetKeys));
+ this.comment = comment;
}
public String getCatalogName() {
return catalogName;
}
- public Set<String> getResetKeys() {
- return resetKeys;
+ public String getComment() {
+ return comment;
}
@Override
public String asSummaryString() {
return String.format(
- "ALTER CATALOG %s\n%s",
- catalogName,
- resetKeys.stream()
- .map(key -> String.format(" RESET '%s'", key))
- .collect(Collectors.joining(",\n")));
+ "ALTER CATALOG %s COMMENT '%s'",
+ catalogName, EncodingUtils.escapeSingleQuotes(comment));
}
@Override
public TableResultInternal execute(Context ctx) {
try {
ctx.getCatalogManager()
- .alterCatalog(catalogName, conf ->
resetKeys.forEach(conf::removeKey));
+ .alterCatalog(catalogName, new
CatalogChange.CatalogCommentChange(comment));
return TableResultImpl.TABLE_RESULT_OK;
} catch (CatalogException e) {
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogOptionsOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogOptionsOperation.java
index e523aa49eb6..77689e8a0f1 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogOptionsOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogOptionsOperation.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.internal.TableResultImpl;
import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.CatalogChange;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import java.util.Collections;
@@ -34,6 +35,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
/** Operation to describe a ALTER CATALOG SET statement. */
@Internal
public class AlterCatalogOptionsOperation implements AlterOperation {
+
private final String catalogName;
private final Map<String, String> properties;
@@ -69,7 +71,9 @@ public class AlterCatalogOptionsOperation implements
AlterOperation {
try {
ctx.getCatalogManager()
.alterCatalog(
- catalogName, conf ->
conf.addAll(Configuration.fromMap(properties)));
+ catalogName,
+ new CatalogChange.CatalogConfigurationChange(
+ conf ->
conf.addAll(Configuration.fromMap(properties))));
return TableResultImpl.TABLE_RESULT_OK;
} catch (CatalogException e) {
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogResetOperation.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogResetOperation.java
index b68387def74..01a9b052683 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogResetOperation.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ddl/AlterCatalogResetOperation.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.internal.TableResultImpl;
import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.catalog.CatalogChange;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import java.util.Collections;
@@ -33,6 +34,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
/** Operation to describe an ALTER CATALOG RESET statement. */
@Internal
public class AlterCatalogResetOperation implements AlterOperation {
+
private final String catalogName;
private final Set<String> resetKeys;
@@ -63,7 +65,10 @@ public class AlterCatalogResetOperation implements
AlterOperation {
public TableResultInternal execute(Context ctx) {
try {
ctx.getCatalogManager()
- .alterCatalog(catalogName, conf ->
resetKeys.forEach(conf::removeKey));
+ .alterCatalog(
+ catalogName,
+ new CatalogChange.CatalogConfigurationChange(
+ conf ->
resetKeys.forEach(conf::removeKey)));
return TableResultImpl.TABLE_RESULT_OK;
} catch (CatalogException e) {
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
index 632387a0ebc..5dac3b1add9 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
@@ -396,6 +396,35 @@ class CatalogManagerTest {
assertEquals(
"comment for catalog",
catalogManager.getCatalogDescriptor("cat_comment").get().getComment().get());
+ assertThat(catalogManager.getCatalog("cat_comment")).isPresent();
+ assertThat(catalogManager.getCatalogDescriptor("cat_comment"))
+ .isPresent()
+ .hasValueSatisfying(
+ descriptor ->
+ assertThat(descriptor.getComment())
+ .isPresent()
+ .hasValueSatisfying(
+ comment ->
+ assertEquals(
+ "comment for
catalog", comment)));
+
+ catalogManager.alterCatalog(
+ "cat_comment",
+ new CatalogChange.CatalogConfigurationChange(
+ conf -> conf.setString("default-database", "db")));
+ catalogManager.alterCatalog(
+ "cat_comment", new CatalogChange.CatalogCommentChange("new
comment"));
+ assertThat(catalogManager.getCatalogDescriptor("cat_comment"))
+ .isPresent()
+ .hasValueSatisfying(
+ descriptor -> {
+ assertThat(descriptor.getConfiguration().toMap())
+ .containsEntry("default-database", "db");
+ assertThat(descriptor.getComment())
+ .isPresent()
+ .hasValueSatisfying(
+ comment -> assertEquals("new
comment", comment));
+ });
assertTrue(catalogManager.listCatalogs().contains("cat1"));
assertTrue(catalogManager.listCatalogs().contains("cat2"));
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogChange.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogChange.java
new file mode 100644
index 00000000000..d446e91258f
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogChange.java
@@ -0,0 +1,75 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+
+import java.util.function.Consumer;
+
+/** {@link CatalogChange} represents the modification of the catalog. */
+@Internal
+public interface CatalogChange {
+
+ /** Generate a new CatalogDescriptor after applying the change to the
given descriptor. */
+ CatalogDescriptor applyChange(CatalogDescriptor descriptor);
+
+ //
--------------------------------------------------------------------------------------------
+ // Option Change
+ //
--------------------------------------------------------------------------------------------
+
+ /** A catalog change to modify the catalog configuration. */
+ @Internal
+ class CatalogConfigurationChange implements CatalogChange {
+
+ private final Consumer<Configuration> configUpdater;
+
+ public CatalogConfigurationChange(Consumer<Configuration>
configUpdater) {
+ this.configUpdater = configUpdater;
+ }
+
+ @Override
+ public CatalogDescriptor applyChange(CatalogDescriptor descriptor) {
+ Configuration conf = descriptor.getConfiguration();
+ configUpdater.accept(conf);
+ return CatalogDescriptor.of(
+ descriptor.getCatalogName(), conf,
descriptor.getComment().orElse(null));
+ }
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // Comment Change
+ //
--------------------------------------------------------------------------------------------
+
+ /** A catalog change to modify the comment. */
+ @Internal
+ class CatalogCommentChange implements CatalogChange {
+
+ private final String newComment;
+
+ public CatalogCommentChange(String newComment) {
+ this.newComment = newComment;
+ }
+
+ @Override
+ public CatalogDescriptor applyChange(CatalogDescriptor descriptor) {
+ return descriptor.setComment(newComment);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogDescriptor.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogDescriptor.java
index f984203672d..38e63d680f9 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogDescriptor.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogDescriptor.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.catalog;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.Configuration;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.Optional;
@@ -59,6 +60,10 @@ public class CatalogDescriptor {
return Optional.ofNullable(comment);
}
+ public CatalogDescriptor setComment(@Nonnull String comment) {
+ return new CatalogDescriptor(catalogName, configuration, comment);
+ }
+
private CatalogDescriptor(
String catalogName, Configuration configuration, @Nullable String
comment) {
this.catalogName = catalogName;
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterCatalogCommentConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterCatalogCommentConverter.java
new file mode 100644
index 00000000000..fced0bc5fe9
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlAlterCatalogCommentConverter.java
@@ -0,0 +1,40 @@
+/*
+ * 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.operations.converters;
+
+import org.apache.flink.sql.parser.ddl.SqlAlterCatalogComment;
+import org.apache.flink.table.operations.Operation;
+import org.apache.flink.table.operations.ddl.AlterCatalogCommentOperation;
+
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.util.NlsString;
+
+/** A converter for {@link SqlAlterCatalogComment}. */
+public class SqlAlterCatalogCommentConverter implements
SqlNodeConverter<SqlAlterCatalogComment> {
+
+ @Override
+ public Operation convertSqlNode(
+ SqlAlterCatalogComment sqlAlterCatalogComment, ConvertContext
context) {
+ return new AlterCatalogCommentOperation(
+ sqlAlterCatalogComment.catalogName(),
+ ((SqlCharStringLiteral) sqlAlterCatalogComment.getComment())
+ .getValueAs(NlsString.class)
+ .getValue());
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
index 5c6fbc8a220..1aeddd6a128 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConverters.java
@@ -41,6 +41,7 @@ public class SqlNodeConverters {
register(new SqlCreateCatalogConverter());
register(new SqlAlterCatalogOptionsConverter());
register(new SqlAlterCatalogResetConverter());
+ register(new SqlAlterCatalogCommentConverter());
register(new SqlCreateViewConverter());
register(new SqlAlterViewRenameConverter());
register(new SqlAlterViewPropertiesConverter());
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
index f854442bce4..969a3f0ef3b 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java
@@ -52,6 +52,7 @@ import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.SinkModifyOperation;
import org.apache.flink.table.operations.SourceQueryOperation;
import org.apache.flink.table.operations.ddl.AddPartitionsOperation;
+import org.apache.flink.table.operations.ddl.AlterCatalogCommentOperation;
import org.apache.flink.table.operations.ddl.AlterCatalogOptionsOperation;
import org.apache.flink.table.operations.ddl.AlterCatalogResetOperation;
import org.apache.flink.table.operations.ddl.AlterDatabaseOperation;
@@ -145,6 +146,20 @@ public class SqlDdlToOperationConverterTest extends
SqlNodeToOperationConversion
assertThatThrownBy(() -> parse("ALTER CATALOG cat2 RESET ()"))
.isInstanceOf(ValidationException.class)
.hasMessageContaining("ALTER CATALOG RESET does not support
empty key");
+
+ // test alter catalog comment
+ operation = parse("ALTER CATALOG cat2 COMMENT 'comment for catalog
''cat2'''");
+ assertThat(operation)
+ .isInstanceOf(AlterCatalogCommentOperation.class)
+
.asInstanceOf(InstanceOfAssertFactories.type(AlterCatalogCommentOperation.class))
+ .extracting(
+ AlterCatalogCommentOperation::getCatalogName,
+ AlterCatalogCommentOperation::asSummaryString,
+ AlterCatalogCommentOperation::getComment)
+ .containsExactly(
+ "cat2",
+ "ALTER CATALOG cat2 COMMENT 'comment for catalog
''cat2'''",
+ "comment for catalog 'cat2'");
}
@Test
