diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowDatabasesOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowDatabasesOperation.java
index 11a6050888150f..e6104652095829 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowDatabasesOperation.java
+++ b/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
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowProceduresOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowProceduresOperation.java
index 29d259090da8b6..aaa7578cabedfd 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowProceduresOperation.java
+++ b/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);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/AbstractSqlShowConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/AbstractSqlShowConverter.java
index 2ba63255aea278..b328f5bd4d54ef 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/AbstractSqlShowConverter.java
+++ b/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);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowFunctionsConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowFunctionsConverter.java
index 1b52548b9b998e..910ae0466539a5 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowFunctionsConverter.java
+++ b/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) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowProcedureConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowProcedureConverter.java
index 8aa1c4952eb289..40e084deee92e5 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowProcedureConverter.java
+++ b/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);
     }
 }
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowTablesConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowTablesConverter.java
index c1f3ecdd6fdad6..d5fb73e4d9f522 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowTablesConverter.java
+++ b/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);
     }
 }
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowViewsConverter.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowViewsConverter.java
index 64399c2f46bf34..0d8a188babee2d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlShowViewsConverter.java
+++ b/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);
     }
 }
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlOtherOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlOtherOperationConverterTest.java
index b088c6ea4d422e..235d8d2a79050f 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlOtherOperationConverterTest.java
+++ b/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