fix(query): create view check unknown table

src/query/service/src/interpreters/interpreter_view_create.rs

@@ -21,6 +21,7 @@ use common_meta_app::schema::CreateTableReq;
 use common_meta_app::schema::TableMeta;
 use common_meta_app::schema::TableNameIdent;
 use common_sql::plans::CreateViewPlan;
+use common_sql::plans::Plan;
 use common_sql::Planner;
 use common_storages_view::view_table::QUERY;
 use common_storages_view::view_table::VIEW_ENGINE;
@@ -70,13 +71,38 @@ impl Interpreter for CreateViewInterpreter {
 impl CreateViewInterpreter {
     async fn create_view(&self) -> Result<PipelineBuildResult> {
         let catalog = self.ctx.get_catalog(&self.plan.catalog)?;
+        let tenant = self.ctx.get_tenant();
+        let table_function = catalog.list_table_functions();
         let mut options = BTreeMap::new();
+        let mut planner = Planner::new(self.ctx.clone());
+        let (plan, _) = planner.plan_sql(&self.plan.subquery.clone()).await?;
+        match plan.clone() {
+            Plan::Query { metadata, .. } => {
+                let metadata = metadata.read().clone();
+                for table in metadata.tables() {
+                    let database_name = table.database();
+                    let table_name = table.name();
+                    if !catalog
+                        .exists_table(tenant.as_str(), database_name, table_name)
+                        .await?
+                        && !table_function.contains(&table_name.to_string())
+                    {
+                        return Err(common_exception::ErrorCode::UnknownTable(format!(
+                            "VIEW QUERY: {}.{} not exists",
+                            database_name, table_name,
+                        )));
+                    }
+                }
+            }
+            _ => {
+                // This logic will never be used, because of QUERY parse as query
+                return Err(ErrorCode::Unimplemented("create view only support Query"));
+            }
+        }
 
         let subquery = if self.plan.column_names.is_empty() {
             self.plan.subquery.clone()
         } else {
-            let mut planner = Planner::new(self.ctx.clone());
-            let (plan, _) = planner.plan_sql(&self.plan.subquery.clone()).await?;
             if plan.schema().fields().len() != self.plan.column_names.len() {
                 return Err(ErrorCode::BadDataArrayLength(format!(
                     "column name length mismatch, expect {}, got {}",

src/query/storages/hive/hive/src/hive_catalog.rs

@@ -409,7 +409,7 @@ impl Catalog for HiveCatalog {
 
     // List all table functions' names.
     fn list_table_functions(&self) -> Vec<String> {
-        unimplemented!()
+        vec![]
     }
 
     // Get table engines

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

@@ -311,7 +311,7 @@ impl Catalog for IcebergCatalog {
 
     // List all table functions' names.
     fn list_table_functions(&self) -> Vec<String> {
-        unimplemented!()
+        vec![]
     }
 
     fn as_any(&self) -> &dyn Any {

tests/sqllogictests/suites/base/05_ddl/05_0019_ddl_create_view

@@ -121,29 +121,11 @@ drop view if exists loop_view2;
 statement ok
 drop view if exists loop_view3;
 
-statement ok
+statement error 1025
 create view loop_view1 as select * from loop_view3;
 
-statement ok
+statement error 1025
 create view loop_view2 as select * from loop_view1;
 
-statement ok
+statement error 1025
 create view loop_view3 as select * from loop_view2;
-
-statement error 1001
-select * from loop_view1;
-
-statement error 1001
-select * from loop_view2;
-
-statement error 1001
-select * from loop_view2;
-
-statement ok
-drop view loop_view1;
-
-statement ok
-drop view loop_view2;
-
-statement ok
-drop view loop_view3;