fix: non-default catalog should not call get_table_function

src/query/service/src/sessions/query_ctx.rs

@@ -80,6 +80,7 @@ use databend_common_meta_app::principal::UserInfo;
 use databend_common_meta_app::principal::UserPrivilegeType;
 use databend_common_meta_app::principal::COPY_MAX_FILES_COMMIT_MSG;
 use databend_common_meta_app::principal::COPY_MAX_FILES_PER_COMMIT;
+use databend_common_meta_app::schema::CatalogType;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::storage::StorageParams;
@@ -191,26 +192,38 @@ impl QueryContext {
             .shared
             .catalog_manager
             .build_catalog(table_info.catalog_info.clone(), self.session_state())?;
-        match table_args {
-            None => {
+
+        let is_default = catalog.info().catalog_type() == CatalogType::Default;
+        match (table_args, is_default) {
+            (Some(table_args), true) => {
+                let default_catalog = self
+                    .shared
+                    .catalog_manager
+                    .get_default_catalog(self.session_state())?;
+                let table_function =
+                    default_catalog.get_table_function(&table_info.name, table_args)?;
+                Ok(table_function.as_table())
+            }
+            (Some(_), false) => Err(ErrorCode::InvalidArgument(
+                "request table args inside non-default catalog is invalid",
+            )),
+            // Load table first, if not found, try to load table function.
+            (None, true) => {
                 let table = catalog.get_table_by_info(table_info);
                 if table.is_err() {
-                    let table_function = catalog
-                        .get_table_function(&table_info.name, TableArgs::new_positioned(vec![]));
+                    let Ok(table_function) = catalog
+                        .get_table_function(&table_info.name, TableArgs::new_positioned(vec![]))
+                    else {
+                        // Returns the table error if the table function failed to load.
+                        return table;
+                    };
 
-                    if table_function.is_err() {
-                        table
-                    } else {
-                        Ok(table_function?.as_table())
-                    }
+                    Ok(table_function.as_table())
                 } else {
                     table
                 }
             }
-
-            Some(table_args) => Ok(catalog
-                .get_table_function(&table_info.name, table_args)?
-                .as_table()),
+            (None, false) => catalog.get_table_by_info(table_info),
         }
     }
 

src/query/storages/iceberg/src/catalog.rs

@@ -262,12 +262,6 @@ impl Catalog for IcebergCatalog {
     }
 
     fn get_table_by_info(&self, table_info: &TableInfo) -> Result<Arc<dyn Table>> {
-        if table_info.meta.storage_params.is_none() {
-            return Err(ErrorCode::BadArguments(
-                "table storage params not set, this is not a valid table info for iceberg table",
-            ));
-        }
-
         let table: Arc<dyn Table> = IcebergTable::try_create(table_info.clone())?.into();
 
         Ok(table)