fix: use the search path when show tables (#16891)

Signed-off-by: tabVersion <[email protected]>

e2e_test/ddl/show.slt

@@ -215,3 +215,37 @@ tablespace character varying false NULL
 hasindexes boolean false NULL
 ispopulated boolean false NULL
 definition character varying false NULL
+
+# show tables according to search path
+
+statement ok
+create schema show_another_schema;
+
+statement ok
+create table show_another_schema.t_another (v1 int);
+
+statement ok
+create table t4 (v1 int);
+
+query T
+show tables;
+----
+t4
+
+statement ok
+set search_path to show_another_schema, public;
+
+query T
+show tables;
+----
+t_another
+t4
+
+statement ok
+drop table t4;
+
+statement ok
+drop table show_another_schema.t_another;
+
+statement ok
+drop schema show_another_schema;

src/frontend/src/handler/show.rs

@@ -21,6 +21,7 @@ use pgwire::pg_response::{PgResponse, StatementType};
 use pgwire::pg_server::Session;
 use risingwave_common::bail_not_implemented;
 use risingwave_common::catalog::{ColumnCatalog, ColumnDesc, DEFAULT_SCHEMA_NAME};
+use risingwave_common::session_config::{SearchPath, USER_NAME_WILD_CARD};
 use risingwave_common::types::{DataType, Fields, Timestamptz};
 use risingwave_common::util::addr::HostAddr;
 use risingwave_connector::source::kafka::PRIVATELINK_CONNECTION;
@@ -105,6 +106,28 @@ fn schema_or_default(schema: &Option<Ident>) -> String {
         .map_or_else(|| DEFAULT_SCHEMA_NAME.to_string(), |s| s.real_value())
 }
 
+fn schema_or_search_path(
+    session: &Arc<SessionImpl>,
+    schema: &Option<Ident>,
+    search_path: &SearchPath,
+) -> Vec<String> {
+    if let Some(s) = schema {
+        vec![s.real_value()]
+    } else {
+        search_path
+            .real_path()
+            .iter()
+            .map(|s| {
+                if s.eq(USER_NAME_WILD_CARD) {
+                    session.auth_context().user_name.clone()
+                } else {
+                    s.to_string()
+                }
+            })
+            .collect()
+    }
+}
+
 #[derive(Fields)]
 #[fields(style = "Title Case")]
 struct ShowObjectRow {
@@ -253,12 +276,22 @@ pub async fn handle_show_object(
 
     let names = match command {
         // If not include schema name, use default schema name
-        ShowObject::Table { schema } => catalog_reader
-            .read_guard()
-            .get_schema_by_name(session.database(), &schema_or_default(&schema))?
-            .iter_table()
-            .map(|t| t.name.clone())
-            .collect(),
+        ShowObject::Table { schema } => {
+            let search_path = session.config().search_path();
+            let mut table_names_in_schema = vec![];
+            for schema in schema_or_search_path(&session, &schema, &search_path) {
+                // If the schema is not found, skip it
+                if let Ok(schema_catalog) = catalog_reader
+                    .read_guard()
+                    .get_schema_by_name(session.database(), schema.as_ref())
+                {
+                    table_names_in_schema
+                        .extend(schema_catalog.iter_table().map(|t| t.name.clone()));
+                }
+            }
+
+            table_names_in_schema
+        }
         ShowObject::InternalTable { schema } => catalog_reader
             .read_guard()
             .get_schema_by_name(session.database(), &schema_or_default(&schema))?