chore: optimize show tables with multi catalogs (#17632)

* chore: optimize show tables

* refactor iceberg show tables

* optimize show databases

* refactor: refactor show tables optimize in tables_table.rs

* fix conversation

* use Option replace init db_name

Author: TCeason

src/query/storages/system/src/databases_table.rs

@@ -27,9 +27,11 @@ use databend_common_expression::types::TimestampType;
 use databend_common_expression::types::UInt64Type;
 use databend_common_expression::utils::FromData;
 use databend_common_expression::DataBlock;
+use databend_common_expression::Scalar;
 use databend_common_expression::TableDataType;
 use databend_common_expression::TableField;
 use databend_common_expression::TableSchemaRefExt;
+use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_common_meta_app::principal::OwnershipObject;
 use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
 use databend_common_meta_app::schema::TableIdent;
@@ -41,6 +43,7 @@ use log::warn;
 
 use crate::table::AsyncOneBlockSystemTable;
 use crate::table::AsyncSystemTable;
+use crate::util::find_eq_filter;
 
 pub type DatabasesTableWithHistory = DatabasesTable<true>;
 pub type DatabasesTableWithoutHistory = DatabasesTable<false>;
@@ -98,17 +101,40 @@ where DatabasesTable<WITH_HISTORY>: HistoryAware
     async fn get_full_data(
         &self,
         ctx: Arc<dyn TableContext>,
-        _push_downs: Option<PushDownInfo>,
+        push_downs: Option<PushDownInfo>,
     ) -> Result<DataBlock> {
         let tenant = ctx.get_tenant();
 
-        let catalogs = CatalogManager::instance();
-        let catalogs: Vec<(String, Arc<dyn Catalog>)> = catalogs
-            .list_catalogs(&tenant, ctx.session_state())
-            .await?
-            .iter()
-            .map(|e| (e.name(), e.clone()))
-            .collect();
+        // Check filters (catalog name)
+        let mut filter_catalog_name = None;
+        if let Some(push_downs) = push_downs {
+            if let Some(filter) = push_downs.filters.as_ref().map(|f| &f.filter) {
+                let expr = filter.as_expr(&BUILTIN_FUNCTIONS);
+                find_eq_filter(&expr, &mut |col_name, scalar| {
+                    if col_name == "catalog" {
+                        if let Scalar::String(catalog) = scalar {
+                            filter_catalog_name = Some(catalog.clone());
+                        }
+                    }
+                    Ok(())
+                });
+            }
+        }
+
+        let catalogs = if let Some(filter_catalog_name) = filter_catalog_name {
+            let mut res = vec![];
+            let ctl = ctx.get_catalog(&filter_catalog_name).await?;
+            res.push((filter_catalog_name, ctl));
+            res
+        } else {
+            let catalogs = CatalogManager::instance();
+            catalogs
+                .list_catalogs(&tenant, ctx.session_state())
+                .await?
+                .iter()
+                .map(|e| (e.name(), e.clone()))
+                .collect()
+        };
 
         let user_api = UserApiProvider::instance();
         let mut catalog_names = vec![];