Address feedback

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowDatabasesOperation.java

@@ -56,8 +56,7 @@ public ShowDatabasesOperation() {
     protected Collection<String> retrieveDataForTableResult(Context ctx) {
         final CatalogManager catalogManager = ctx.getCatalogManager();
         final String catalogName = catalogManager.qualifyCatalog(this.catalogName);
-        String cName = catalogName == null ? catalogManager.getCurrentCatalog() : catalogName;
-        return catalogManager.getCatalogOrThrowException(cName).listDatabases();
+        return catalogManager.getCatalogOrThrowException(catalogName).listDatabases();
     }
 
     @Override

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowProceduresOperation.java

@@ -65,7 +65,7 @@ protected Collection<String> retrieveDataForTableResult(Context ctx) {
             if (preposition == null) {
                 // it's to show current_catalog.current_database
                 return catalogManager
-                        .getCatalogOrError(catalogManager.getCurrentCatalog())
+                        .getCatalogOrThrowException(catalogManager.getCurrentCatalog())
                         .listProcedures(catalogManager.getCurrentDatabase());
             } else {
                 Catalog catalog = catalogManager.getCatalogOrThrowException(catalogName);

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/AbstractSqlShowConverter.java

@@ -25,30 +25,36 @@
 import org.apache.flink.table.operations.utils.LikeType;
 import org.apache.flink.table.operations.utils.ShowLikeOperator;
 
+import javax.annotation.Nullable;
+
 import java.util.List;
-import java.util.function.Function;
 
 public abstract class AbstractSqlShowConverter<T extends SqlShowCall>
         implements SqlNodeConverter<T> {
 
-    protected Operation convertShowOperation(
-            T sqlShowCall, Function<List<String>, String> msg, ConvertContext context) {
+    protected Operation convertShowOperation(T sqlShowCall, ConvertContext context) {
         final ShowLikeOperator likeOp = getLikeOp(sqlShowCall);
         if (sqlShowCall.getPreposition() == null) {
             return getOperationWithoutPrep(sqlShowCall, likeOp);
         }
-        List<String> fullDatabaseName = sqlShowCall.getSqlIdentifierNameList();
-        if (fullDatabaseName.size() > 2) {
-            throw new ValidationException(msg.apply(fullDatabaseName));
+        List<String> sqlIdentifierNameList = sqlShowCall.getSqlIdentifierNameList();
+        if (sqlIdentifierNameList.size() > 2) {
+            throw new ValidationException(
+                    String.format(
+                            "%s from/in identifier [ %s ] format error, it should be [catalog_name.]database_name.",
+                            sqlShowCall.getOperator().getName(),
+                            String.join(".", sqlIdentifierNameList)));
         }
         CatalogManager catalogManager = context.getCatalogManager();
         String catalogName =
-                fullDatabaseName.size() == 1
+                sqlIdentifierNameList.size() == 1
                         ? catalogManager.getCurrentCatalog()
-                        : fullDatabaseName.get(0);
+                        : sqlIdentifierNameList.get(0);
 
         String databaseName =
-                fullDatabaseName.size() == 1 ? fullDatabaseName.get(0) : fullDatabaseName.get(1);
+                sqlIdentifierNameList.size() == 1
+                        ? sqlIdentifierNameList.get(0)
+                        : sqlIdentifierNameList.get(1);
         return getOperation(
                 sqlShowCall, catalogName, databaseName, sqlShowCall.getPreposition(), likeOp);
     }
@@ -63,10 +69,10 @@ public ShowLikeOperator getLikeOp(SqlShowCall sqlShowCall) {
 
     public abstract Operation getOperation(
             T sqlShowCall,
-            String catalogName,
-            String databaseName,
-            String prep,
-            ShowLikeOperator likeOp);
+            @Nullable String catalogName,
+            @Nullable String databaseName,
+            @Nullable String prep,
+            @Nullable ShowLikeOperator likeOp);
 
     @Override
     public abstract Operation convertSqlNode(T node, ConvertContext context);

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowFunctionsConverter.java

@@ -47,13 +47,7 @@ public Operation getOperation(
 
     @Override
     public Operation convertSqlNode(SqlShowFunctions sqlShowFunctions, ConvertContext context) {
-        return convertShowOperation(
-                sqlShowFunctions,
-                sqlIdentifierNameList ->
-                        String.format(
-                                "Show functions from/in identifier [ %s ] format error, it should be [catalog_name.]database_name.",
-                                String.join(".", sqlIdentifierNameList)),
-                context);
+        return convertShowOperation(sqlShowFunctions, context);
     }
 
     private static FunctionScope getFunctionScope(SqlShowFunctions sqlShowFunctions) {

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowProcedureConverter.java

@@ -44,12 +44,6 @@ public Operation getOperation(
 
     @Override
     public Operation convertSqlNode(SqlShowProcedures sqlShowProcedures, ConvertContext context) {
-        return convertShowOperation(
-                sqlShowProcedures,
-                sqlIdentifierNameList ->
-                        String.format(
-                                "Show procedures from/in identifier [ %s ] format error, it should be [catalog_name.]database_name.",
-                                String.join(".", sqlIdentifierNameList)),
-                context);
+        return convertShowOperation(sqlShowProcedures, context);
     }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowTablesConverter.java

@@ -41,12 +41,6 @@ public Operation getOperation(
 
     @Override
     public Operation convertSqlNode(SqlShowTables sqlShowTables, ConvertContext context) {
-        return convertShowOperation(
-                sqlShowTables,
-                sqlIdentifierNameList ->
-                        String.format(
-                                "show tables from/in identifier [ %s ] format error",
-                                String.join(".", sqlIdentifierNameList)),
-                context);
+        return convertShowOperation(sqlShowTables, context);
     }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowViewsConverter.java

@@ -41,12 +41,6 @@ public Operation getOperation(
 
     @Override
     public Operation convertSqlNode(SqlShowViews sqlShowViews, ConvertContext context) {
-        return convertShowOperation(
-                sqlShowViews,
-                sqlIdentifierNameList ->
-                        String.format(
-                                "show views from/in identifier [ %s ] format error",
-                                String.join(".", sqlIdentifierNameList)),
-                context);
+        return convertShowOperation(sqlShowViews, context);
     }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlOtherOperationConverterTest.java

@@ -310,15 +310,6 @@ private static Stream<Arguments> inputForShowFunctionsTest() {
                         "SHOW FUNCTIONS IN builtin.db1"));
     }
 
-    @Test
-    void testShowFunctionsForFailedCase() {
-        // test fail case
-        assertThatThrownBy(() -> parse("show functions in cat.db.t"))
-                .isInstanceOf(ValidationException.class)
-                .hasMessage(
-                        "Show functions from/in identifier [ cat.db.t ] format error, it should be [catalog_name.]database_name.");
-    }
-
     @Test
     void testShowDatabasesFailCase() {
         assertThatThrownBy(() -> parse("show databases in db.t"))
@@ -394,13 +385,33 @@ private static Stream<Arguments> inputForShowProceduresTest() {
                         "SHOW procedures", new ShowProceduresOperation(null, null, null, null)));
     }
 
-    @Test
-    void testShowProceduresFailCase() {
+    @ParameterizedTest
+    @MethodSource("argsForTestShowFailedCase")
+    void testShowProceduresFailCase(String sql, String expectedErrorMsg) {
         // test fail case
-        assertThatThrownBy(() -> parse("SHOW procedures in cat.db.t"))
+        assertThatThrownBy(() -> parse(sql))
                 .isInstanceOf(ValidationException.class)
-                .hasMessage(
-                        "Show procedures from/in identifier [ cat.db.t ] format error, it should be [catalog_name.]database_name.");
+                .hasMessage(expectedErrorMsg);
+    }
+
+    private static Stream<Arguments> argsForTestShowFailedCase() {
+        return Stream.of(
+                Arguments.of(
+                        "SHOW procedures in cat.db.t",
+                        "SHOW PROCEDURES from/in identifier [ cat.db.t ] format error,"
+                                + " it should be [catalog_name.]database_name."),
+                Arguments.of(
+                        "SHOW Views in cat.db1.t2",
+                        "SHOW VIEWS from/in identifier [ cat.db1.t2 ] format error,"
+                                + " it should be [catalog_name.]database_name."),
+                Arguments.of(
+                        "SHOW functions in cat.db3.t5",
+                        "SHOW FUNCTIONS from/in identifier [ cat.db3.t5 ] format error,"
+                                + " it should be [catalog_name.]database_name."),
+                Arguments.of(
+                        "SHOW tables in cat1.db3.t2",
+                        "SHOW TABLES from/in identifier [ cat1.db3.t2 ] format error,"
+                                + " it should be [catalog_name.]database_name."));
     }
 
     @Test