From 8b94eeb472e1a19bcec68f44aa2c44f551c08967 Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Thu, 25 Aug 2022 13:50:37 -0400
Subject: [PATCH 1/8] add initial support for 'drop view'

---
 datafusion/core/src/execution/context.rs      | 104 ++++++++++++------
 datafusion/core/src/physical_plan/planner.rs  |   2 +-
 datafusion/expr/src/logical_plan/mod.rs       |   8 +-
 datafusion/expr/src/logical_plan/plan.rs      |  28 ++++-
 datafusion/expr/src/utils.rs                  |   1 +
 .../optimizer/src/common_subexpr_eliminate.rs |   1 +
 .../optimizer/src/projection_push_down.rs     |   1 +
 datafusion/proto/src/logical_plan.rs          |   3 +
 datafusion/sql/src/planner.rs                 |  45 +++++---
 9 files changed, 135 insertions(+), 58 deletions(-)

diff --git a/datafusion/core/src/execution/context.rs b/datafusion/core/src/execution/context.rs
index a0dee62de78f..4ce48a064dbc 100644
--- a/datafusion/core/src/execution/context.rs
+++ b/datafusion/core/src/execution/context.rs
@@ -102,7 +102,8 @@ use crate::variable::{VarProvider, VarType};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use datafusion_common::ScalarValue;
-use datafusion_expr::TableSource;
+use datafusion_expr::logical_plan::DropView;
+use datafusion_expr::{TableSource, TableType};
 use datafusion_optimizer::decorrelate_scalar_subquery::DecorrelateScalarSubquery;
 use datafusion_optimizer::decorrelate_where_exists::DecorrelateWhereExists;
 use datafusion_optimizer::decorrelate_where_in::DecorrelateWhereIn;
@@ -269,10 +270,7 @@ impl SessionContext {
                 };
                 let table = self.table(name.as_str());
                 match (if_not_exists, table) {
-                    (true, Ok(_)) => {
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
-                    }
+                    (true, Ok(_)) => self.return_empty_dataframe(),
                     (_, Err(_)) => {
                         // TODO make schema in CreateExternalTable optional instead of empty
                         let provided_schema = if schema.fields().is_empty() {
@@ -294,8 +292,7 @@ impl SessionContext {
                             provided_schema,
                         )
                         .await?;
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
+                        self.return_empty_dataframe()
                     }
                     (false, Ok(_)) => Err(DataFusionError::Execution(format!(
                         "Table '{:?}' already exists",
@@ -313,10 +310,7 @@ impl SessionContext {
                 let table = self.table(name.as_str());
 
                 match (if_not_exists, or_replace, table) {
-                    (true, false, Ok(_)) => {
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
-                    }
+                    (true, false, Ok(_)) => self.return_empty_dataframe(),
                     (false, true, Ok(_)) => {
                         self.deregister_table(name.as_str())?;
                         let plan = self.optimize(&input)?;
@@ -371,16 +365,14 @@ impl SessionContext {
                             Arc::new(ViewTable::try_new((*input).clone(), definition)?);
 
                         self.register_table(name.as_str(), table)?;
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
+                        self.return_empty_dataframe()
                     }
                     (_, Err(_)) => {
                         let table =
                             Arc::new(ViewTable::try_new((*input).clone(), definition)?);
 
                         self.register_table(name.as_str(), table)?;
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
+                        self.return_empty_dataframe()
                     }
                     (false, Ok(_)) => Err(DataFusionError::Execution(format!(
                         "Table '{:?}' already exists",
@@ -392,15 +384,44 @@ impl SessionContext {
             LogicalPlan::DropTable(DropTable {
                 name, if_exists, ..
             }) => {
-                let returned = self.deregister_table(name.as_str())?;
-                if !if_exists && returned.is_none() {
+                let tp = self.get_table_provider(name.as_str());
+                // if table exists and is a table, deregister it
+                if let Some(table_provider) = tp {
+                    if table_provider.table_type() == TableType::Base {
+                        self.deregister_table(name.as_str())?;
+                        return self.return_empty_dataframe();
+                    }
+                }
+                // table doesn't exist or isn't a table, so either return quietly or show error
+                if if_exists {
+                    self.return_empty_dataframe()
+                } else {
                     Err(DataFusionError::Execution(format!(
-                        "Memory table {:?} doesn't exist.",
+                        "Table {:?} doesn't exist.",
                         name
                     )))
+                }
+            }
+
+            LogicalPlan::DropView(DropView {
+                name, if_exists, ..
+            }) => {
+                let tp = self.get_table_provider(name.as_str());
+                // if view exists and is a view, deregister it
+                if let Some(table_provider) = tp {
+                    if table_provider.table_type() == TableType::View {
+                        self.deregister_table(name.as_str())?;
+                        return self.return_empty_dataframe();
+                    }
+                }
+                // view doesn't exist or isn't a view, so either return quietly or show error
+                if if_exists {
+                    self.return_empty_dataframe()
                 } else {
-                    let plan = LogicalPlanBuilder::empty(false).build()?;
-                    Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
+                    Err(DataFusionError::Execution(format!(
+                        "View {:?} doesn't exist.",
+                        name
+                    )))
                 }
             }
             LogicalPlan::CreateCatalogSchema(CreateCatalogSchema {
@@ -429,15 +450,11 @@ impl SessionContext {
                 let schema = catalog.schema(schema_name);
 
                 match (if_not_exists, schema) {
-                    (true, Some(_)) => {
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
-                    }
+                    (true, Some(_)) => self.return_empty_dataframe(),
                     (true, None) | (false, None) => {
                         let schema = Arc::new(MemorySchemaProvider::new());
                         catalog.register_schema(schema_name, schema)?;
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
+                        self.return_empty_dataframe()
                     }
                     (false, Some(_)) => Err(DataFusionError::Execution(format!(
                         "Schema '{:?}' already exists",
@@ -453,18 +470,14 @@ impl SessionContext {
                 let catalog = self.catalog(catalog_name.as_str());
 
                 match (if_not_exists, catalog) {
-                    (true, Some(_)) => {
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
-                    }
+                    (true, Some(_)) => self.return_empty_dataframe(),
                     (true, None) | (false, None) => {
                         let new_catalog = Arc::new(MemoryCatalogProvider::new());
                         self.state
                             .write()
                             .catalog_list
                             .register_catalog(catalog_name, new_catalog);
-                        let plan = LogicalPlanBuilder::empty(false).build()?;
-                        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
+                        self.return_empty_dataframe()
                     }
                     (false, Some(_)) => Err(DataFusionError::Execution(format!(
                         "Catalog '{:?}' already exists",
@@ -477,6 +490,12 @@ impl SessionContext {
         }
     }
 
+    // return an empty dataframe
+    fn return_empty_dataframe(&self) -> Result<Arc<DataFrame>> {
+        let plan = LogicalPlanBuilder::empty(false).build()?;
+        Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
+    }
+
     /// Creates a logical plan.
     ///
     /// This function is intended for internal use and should not be called directly.
@@ -946,6 +965,27 @@ impl SessionContext {
     pub fn state(&self) -> SessionState {
         self.state.read().clone()
     }
+
+    /// Retrieves a TableProvider representing a table previously registered by calling the
+    /// register_table function.
+    ///
+    /// Returns none if no table has been registered with the provided reference.
+    fn get_table_provider<'a>(
+        &self,
+        table_ref: impl Into<TableReference<'a>>,
+    ) -> Option<Arc<dyn TableProvider>> {
+        let table_ref = table_ref.into();
+        let table = self
+            .state
+            .read()
+            .schema_for_ref(table_ref)
+            .ok()?
+            .table(table_ref.table());
+        match table {
+            Some(_) => table,
+            _ => None,
+        }
+    }
 }
 
 impl FunctionRegistry for SessionContext {
diff --git a/datafusion/core/src/physical_plan/planner.rs b/datafusion/core/src/physical_plan/planner.rs
index 658da6a0fef6..dafc013361de 100644
--- a/datafusion/core/src/physical_plan/planner.rs
+++ b/datafusion/core/src/physical_plan/planner.rs
@@ -961,7 +961,7 @@ impl DefaultPhysicalPlanner {
                         "Unsupported logical plan: CreateCatalog".to_string(),
                     ))
                 }
-                | LogicalPlan::CreateMemoryTable(_) | LogicalPlan::DropTable (_) | LogicalPlan::CreateView(_) => {
+                | LogicalPlan::CreateMemoryTable(_) | LogicalPlan::DropTable(_) | LogicalPlan::DropView(_) | LogicalPlan::CreateView(_) => {
                     // Create a dummy exec.
                     Ok(Arc::new(EmptyExec::new(
                         false,
diff --git a/datafusion/expr/src/logical_plan/mod.rs b/datafusion/expr/src/logical_plan/mod.rs
index ff0f5925406e..9917d69a7910 100644
--- a/datafusion/expr/src/logical_plan/mod.rs
+++ b/datafusion/expr/src/logical_plan/mod.rs
@@ -23,10 +23,10 @@ mod plan;
 pub use builder::{table_scan, LogicalPlanBuilder};
 pub use plan::{
     Aggregate, Analyze, CreateCatalog, CreateCatalogSchema, CreateExternalTable,
-    CreateMemoryTable, CreateView, CrossJoin, Distinct, DropTable, EmptyRelation,
-    Explain, Extension, FileType, Filter, Join, JoinConstraint, JoinType, Limit,
-    LogicalPlan, Partitioning, PlanType, PlanVisitor, Projection, Repartition, Sort,
-    StringifiedPlan, Subquery, SubqueryAlias, TableScan, ToStringifiedPlan, Union,
+    CreateMemoryTable, CreateView, CrossJoin, Distinct, DropTable, DropView,
+    EmptyRelation, Explain, Extension, FileType, Filter, Join, JoinConstraint, JoinType,
+    Limit, LogicalPlan, Partitioning, PlanType, PlanVisitor, Projection, Repartition,
+    Sort, StringifiedPlan, Subquery, SubqueryAlias, TableScan, ToStringifiedPlan, Union,
     Values, Window,
 };
 
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 2d5eb46804df..239b4e374eb0 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -86,6 +86,8 @@ pub enum LogicalPlan {
     CreateCatalog(CreateCatalog),
     /// Drops a table.
     DropTable(DropTable),
+    /// Drops a view.
+    DropView(DropView),
     /// Values expression. See
     /// [Postgres VALUES](https://www.postgresql.org/docs/current/queries-values.html)
     /// documentation for more details.
@@ -137,6 +139,7 @@ impl LogicalPlan {
             }
             LogicalPlan::CreateCatalog(CreateCatalog { schema, .. }) => schema,
             LogicalPlan::DropTable(DropTable { schema, .. }) => schema,
+            LogicalPlan::DropView(DropView { schema, .. }) => schema,
         }
     }
 
@@ -193,7 +196,7 @@ impl LogicalPlan {
             | LogicalPlan::CreateView(CreateView { input, .. })
             | LogicalPlan::Filter(Filter { input, .. }) => input.all_schemas(),
             LogicalPlan::Distinct(Distinct { input, .. }) => input.all_schemas(),
-            LogicalPlan::DropTable(_) => vec![],
+            LogicalPlan::DropTable(_) | LogicalPlan::DropView(_) => vec![],
         }
     }
 
@@ -253,6 +256,7 @@ impl LogicalPlan {
             | LogicalPlan::CreateCatalogSchema(_)
             | LogicalPlan::CreateCatalog(_)
             | LogicalPlan::DropTable(_)
+            | LogicalPlan::DropView(_)
             | LogicalPlan::CrossJoin(_)
             | LogicalPlan::Analyze { .. }
             | LogicalPlan::Explain { .. }
@@ -296,7 +300,8 @@ impl LogicalPlan {
             | LogicalPlan::CreateExternalTable(_)
             | LogicalPlan::CreateCatalogSchema(_)
             | LogicalPlan::CreateCatalog(_)
-            | LogicalPlan::DropTable(_) => vec![],
+            | LogicalPlan::DropTable(_)
+            | LogicalPlan::DropView(_) => vec![],
         }
     }
 
@@ -449,7 +454,8 @@ impl LogicalPlan {
             | LogicalPlan::CreateExternalTable(_)
             | LogicalPlan::CreateCatalogSchema(_)
             | LogicalPlan::CreateCatalog(_)
-            | LogicalPlan::DropTable(_) => true,
+            | LogicalPlan::DropTable(_)
+            | LogicalPlan::DropView(_) => true,
         };
         if !recurse {
             return Ok(false);
@@ -920,6 +926,11 @@ impl LogicalPlan {
                     }) => {
                         write!(f, "DropTable: {:?} if not exist:={}", name, if_exists)
                     }
+                    LogicalPlan::DropView(DropView {
+                        name, if_exists, ..
+                    }) => {
+                        write!(f, "DropView: {:?} if not exist:={}", name, if_exists)
+                    }
                     LogicalPlan::Distinct(Distinct { .. }) => {
                         write!(f, "Distinct:")
                     }
@@ -1019,6 +1030,17 @@ pub struct DropTable {
     pub schema: DFSchemaRef,
 }
 
+/// Drops a view.
+#[derive(Clone)]
+pub struct DropView {
+    /// The view name
+    pub name: String,
+    /// If the view exists
+    pub if_exists: bool,
+    /// Dummy schema
+    pub schema: DFSchemaRef,
+}
+
 /// Produces no rows: An empty relation with an empty schema
 #[derive(Clone)]
 pub struct EmptyRelation {
diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs
index 367c722d220a..664181c892f4 100644
--- a/datafusion/expr/src/utils.rs
+++ b/datafusion/expr/src/utils.rs
@@ -515,6 +515,7 @@ pub fn from_plan(
         | LogicalPlan::TableScan { .. }
         | LogicalPlan::CreateExternalTable(_)
         | LogicalPlan::DropTable(_)
+        | LogicalPlan::DropView(_)
         | LogicalPlan::CreateCatalogSchema(_)
         | LogicalPlan::CreateCatalog(_) => {
             // All of these plan types have no inputs / exprs so should not be called
diff --git a/datafusion/optimizer/src/common_subexpr_eliminate.rs b/datafusion/optimizer/src/common_subexpr_eliminate.rs
index 8627b404dce8..20b823929558 100644
--- a/datafusion/optimizer/src/common_subexpr_eliminate.rs
+++ b/datafusion/optimizer/src/common_subexpr_eliminate.rs
@@ -231,6 +231,7 @@ fn optimize(
         | LogicalPlan::CreateCatalogSchema(_)
         | LogicalPlan::CreateCatalog(_)
         | LogicalPlan::DropTable(_)
+        | LogicalPlan::DropView(_)
         | LogicalPlan::Distinct(_)
         | LogicalPlan::Extension { .. } => {
             // apply the optimization to all inputs of the plan
diff --git a/datafusion/optimizer/src/projection_push_down.rs b/datafusion/optimizer/src/projection_push_down.rs
index aa3cdfb4252c..4e2aeac159ae 100644
--- a/datafusion/optimizer/src/projection_push_down.rs
+++ b/datafusion/optimizer/src/projection_push_down.rs
@@ -500,6 +500,7 @@ fn optimize_plan(
         | LogicalPlan::CreateCatalogSchema(_)
         | LogicalPlan::CreateCatalog(_)
         | LogicalPlan::DropTable(_)
+        | LogicalPlan::DropView(_)
         | LogicalPlan::CrossJoin(_)
         | LogicalPlan::Distinct(_)
         | LogicalPlan::Extension { .. } => {
diff --git a/datafusion/proto/src/logical_plan.rs b/datafusion/proto/src/logical_plan.rs
index bcdb9796f17e..ca0487d39c0e 100644
--- a/datafusion/proto/src/logical_plan.rs
+++ b/datafusion/proto/src/logical_plan.rs
@@ -1215,6 +1215,9 @@ impl AsLogicalPlan for LogicalPlanNode {
             LogicalPlan::DropTable(_) => Err(proto_error(
                 "LogicalPlan serde is not yet implemented for DropTable",
             )),
+            LogicalPlan::DropView(_) => Err(proto_error(
+                "LogicalPlan serde is not yet implemented for DropView",
+            )),
         }
     }
 }
diff --git a/datafusion/sql/src/planner.rs b/datafusion/sql/src/planner.rs
index 094b980f1f85..2dc49ef8494f 100644
--- a/datafusion/sql/src/planner.rs
+++ b/datafusion/sql/src/planner.rs
@@ -26,7 +26,7 @@ use datafusion_expr::expr_rewriter::normalize_col_with_schemas;
 use datafusion_expr::logical_plan::{
     Analyze, CreateCatalog, CreateCatalogSchema,
     CreateExternalTable as PlanCreateExternalTable, CreateMemoryTable, CreateView,
-    DropTable, Explain, FileType, JoinType, LogicalPlan, LogicalPlanBuilder,
+    DropTable, DropView, Explain, FileType, JoinType, LogicalPlan, LogicalPlanBuilder,
     Partitioning, PlanType, ToStringifiedPlan,
 };
 use datafusion_expr::utils::{
@@ -245,20 +245,29 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                 schema: Arc::new(DFSchema::empty()),
             })),
             Statement::Drop {
-                object_type: ObjectType::Table,
+                object_type,
                 if_exists,
                 names,
                 cascade: _,
                 purge: _,
-            } =>
-            // We don't support cascade and purge for now.
-            {
-                Ok(LogicalPlan::DropTable(DropTable {
+                // We don't support cascade and purge for now.
+                // nor do we support multiple object names
+            } => match object_type {
+                ObjectType::Table => Ok(LogicalPlan::DropTable(DropTable {
                     name: names.get(0).unwrap().to_string(),
                     if_exists,
                     schema: DFSchemaRef::new(DFSchema::empty()),
-                }))
-            }
+                })),
+                ObjectType::View => Ok(LogicalPlan::DropView(DropView {
+                    name: names.get(0).unwrap().to_string(),
+                    if_exists,
+                    schema: DFSchemaRef::new(DFSchema::empty()),
+                })),
+                _ => Err(DataFusionError::NotImplemented(
+                    "Only `DROP TABLE/VIEW  ...` statement is supported currently"
+                        .to_string(),
+                )),
+            },
 
             Statement::ShowTables {
                 extended,
@@ -4193,7 +4202,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                               QUERY PLAN
     /// ----------------------------------------------------------------------
     /// WindowAgg  (cost=69.83..87.33 rows=1000 width=8)
@@ -4212,7 +4221,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                                     QUERY PLAN
     /// ----------------------------------------------------------------------------------
     /// WindowAgg  (cost=137.16..154.66 rows=1000 width=12)
@@ -4300,7 +4309,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                                     QUERY PLAN
     /// -----------------------------------------------------------------------------------
     /// WindowAgg  (cost=142.16..162.16 rows=1000 width=16)
@@ -4323,7 +4332,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                                        QUERY PLAN
     /// ----------------------------------------------------------------------------------------
     /// WindowAgg  (cost=139.66..172.16 rows=1000 width=24)
@@ -4348,7 +4357,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                                     QUERY PLAN
     /// ----------------------------------------------------------------------------------
     /// WindowAgg  (cost=69.83..117.33 rows=1000 width=24)
@@ -4373,7 +4382,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                                        QUERY PLAN
     /// ----------------------------------------------------------------------------------------
     /// WindowAgg  (cost=139.66..172.16 rows=1000 width=24)
@@ -4402,7 +4411,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                               QUERY PLAN
     /// ----------------------------------------------------------------------
     /// WindowAgg  (cost=69.83..89.83 rows=1000 width=12)
@@ -4422,7 +4431,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                               QUERY PLAN
     /// ----------------------------------------------------------------------
     /// WindowAgg  (cost=69.83..89.83 rows=1000 width=12)
@@ -4442,7 +4451,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                                     QUERY PLAN
     /// ----------------------------------------------------------------------------------
     /// WindowAgg  (cost=142.16..162.16 rows=1000 width=16)
@@ -4466,7 +4475,7 @@ mod tests {
     }
 
     /// psql result
-    /// ```
+    /// ```text
     ///                                  QUERY PLAN
     /// -----------------------------------------------------------------------------
     /// WindowAgg  (cost=69.83..109.83 rows=1000 width=24)

From b6c023eff0f5ac1d338ed402d44d25502f5b6c04 Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Thu, 25 Aug 2022 15:30:23 -0400
Subject: [PATCH 2/8] add tests

---
 datafusion/core/tests/sql/create_drop.rs | 103 +++++++++++++++++++++++
 1 file changed, 103 insertions(+)

diff --git a/datafusion/core/tests/sql/create_drop.rs b/datafusion/core/tests/sql/create_drop.rs
index 1d28f2e3371e..f1fc6cbd8a6b 100644
--- a/datafusion/core/tests/sql/create_drop.rs
+++ b/datafusion/core/tests/sql/create_drop.rs
@@ -112,6 +112,104 @@ async fn drop_table() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn drop_view() -> Result<()> {
+    let ctx =
+        SessionContext::with_config(SessionConfig::new().with_information_schema(true));
+    plan_and_collect(&ctx, "CREATE VIEW v AS SELECT 1").await?;
+    let rb = plan_and_collect(
+        &ctx,
+        "select * from information_schema.tables where table_name = 'v' and table_type = 'VIEW'",
+    )
+    .await?;
+    assert_eq!(rb[0].num_rows(), 1);
+
+    plan_and_collect(&ctx, "DROP VIEW v").await?;
+    let rb = plan_and_collect(
+        &ctx,
+        "select * from information_schema.tables where table_name = 'v' and table_type = 'VIEW'",
+    )
+    .await?;
+    assert!(rb.is_empty());
+    Ok(())
+}
+
+#[tokio::test]
+#[should_panic(expected = "doesn't exist")]
+async fn drop_view_nonexistent() {
+    let ctx = SessionContext::new();
+    ctx.sql("DROP VIEW non_existent_view")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+#[should_panic(expected = "doesn't exist")]
+async fn drop_view_cant_drop_table() {
+    let ctx = SessionContext::new();
+    ctx.sql("CREATE TABLE t AS SELECT 1")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    ctx.sql("DROP VIEW t")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+#[should_panic(expected = "doesn't exist")]
+async fn drop_table_cant_drop_view() {
+    let ctx = SessionContext::new();
+    ctx.sql("CREATE VIEW v AS SELECT 1")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    ctx.sql("DROP TABLE v")
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+async fn drop_view_if_exists() -> Result<()> {
+    let ctx =
+        SessionContext::with_config(SessionConfig::new().with_information_schema(true));
+    plan_and_collect(&ctx, "CREATE VIEW v AS SELECT 1").await?;
+    let rb = plan_and_collect(&ctx, "DROP VIEW IF EXISTS non_existent_view").await?;
+    // make sure we get an empty response back
+    assert!(rb.is_empty());
+
+    let rb = plan_and_collect(
+        &ctx,
+        "select * from information_schema.views where table_name = 'v'",
+    )
+    .await?;
+    // confirm view exists
+    assert_eq!(rb[0].num_rows(), 1);
+
+    plan_and_collect(&ctx, "DROP VIEW IF EXISTS v").await?;
+    let rb = plan_and_collect(
+        &ctx,
+        "select * from information_schema.views where table_name = 'v'",
+    )
+    .await?;
+    // confirm view is gone
+    assert!(rb.is_empty());
+    Ok(())
+}
+
 #[tokio::test]
 async fn csv_query_create_external_table() {
     let ctx = SessionContext::new();
@@ -261,3 +359,8 @@ async fn create_pipe_delimited_csv_table() -> Result<()> {
 
     Ok(())
 }
+
+/// Execute SQL and return results
+async fn plan_and_collect(ctx: &SessionContext, sql: &str) -> Result<Vec<RecordBatch>> {
+    ctx.sql(sql).await?.collect().await
+}

From f84eecc6d0bc2600cbeddfb800a62535a0cab34d Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Thu, 25 Aug 2022 15:41:43 -0400
Subject: [PATCH 3/8] add very basic docs

---
 docs/source/user-guide/sql/ddl.md | 19 ++++++++++++++-----
 1 file changed, 14 insertions(+), 5 deletions(-)

diff --git a/docs/source/user-guide/sql/ddl.md b/docs/source/user-guide/sql/ddl.md
index ee73370e0c25..c5cf29cf3592 100644
--- a/docs/source/user-guide/sql/ddl.md
+++ b/docs/source/user-guide/sql/ddl.md
@@ -94,14 +94,23 @@ CREATE TABLE memtable as select * from valuetable;
 
 ## DROP TABLE
 
-The table can be deleted.
+Removes the table from DataFusion's catalog.
 
-```
-DROP TABLE [ IF EXISTS ] name
-```
+<pre>
+DROP TABLE [ IF EXISTS ] <b><i>table_name</i></b>;
+</pre>
 
 ```sql
 CREATE TABLE users AS VALUES(1,2),(2,3);
-
 DROP TABLE users;
+-- or use 'if exists' to silently ignore if the table doesn't exist
+DROP TABLE IF EXISTS nonexistent_table;
 ```
+
+## DROP VIEW
+
+Removes the view from DataFusion's catalog.
+
+<pre>
+DROP VIEW [ IF EXISTS ] <b><i>view_name</i></b>;
+</pre>

From 947e7137145009eea0ef114f749b892a7577570f Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Thu, 25 Aug 2022 16:16:28 -0400
Subject: [PATCH 4/8] add example to drop view doc

---
 docs/source/user-guide/sql/ddl.md | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/docs/source/user-guide/sql/ddl.md b/docs/source/user-guide/sql/ddl.md
index c5cf29cf3592..3754a8ac0322 100644
--- a/docs/source/user-guide/sql/ddl.md
+++ b/docs/source/user-guide/sql/ddl.md
@@ -109,8 +109,13 @@ DROP TABLE IF EXISTS nonexistent_table;
 
 ## DROP VIEW
 
-Removes the view from DataFusion's catalog.
+Removes the view from DataFusion's catalog.add 
 
 <pre>
 DROP VIEW [ IF EXISTS ] <b><i>view_name</i></b>;
 </pre>
+
+```sql
+-- drop users_v view from the customer_a schema
+DROP VIEW IF EXISTS customer_a.users_v; 
+```

From 5f3f77ee5658d05a2895da4b0c92d18f2021fe31 Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Thu, 25 Aug 2022 16:38:30 -0400
Subject: [PATCH 5/8] ran prettier on the doc

---
 docs/source/user-guide/sql/ddl.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/source/user-guide/sql/ddl.md b/docs/source/user-guide/sql/ddl.md
index 3754a8ac0322..dc3f971171ac 100644
--- a/docs/source/user-guide/sql/ddl.md
+++ b/docs/source/user-guide/sql/ddl.md
@@ -109,7 +109,7 @@ DROP TABLE IF EXISTS nonexistent_table;
 
 ## DROP VIEW
 
-Removes the view from DataFusion's catalog.add 
+Removes the view from DataFusion's catalog.add
 
 <pre>
 DROP VIEW [ IF EXISTS ] <b><i>view_name</i></b>;
@@ -117,5 +117,5 @@ DROP VIEW [ IF EXISTS ] <b><i>view_name</i></b>;
 
 ```sql
 -- drop users_v view from the customer_a schema
-DROP VIEW IF EXISTS customer_a.users_v; 
+DROP VIEW IF EXISTS customer_a.users_v;
 ```

From b90f722579eb2d12cab932ff871af4f1e78f6ad5 Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Thu, 25 Aug 2022 17:27:13 -0400
Subject: [PATCH 6/8] refactored, now done

---
 datafusion/core/src/execution/context.rs | 81 ++++++++++--------------
 1 file changed, 32 insertions(+), 49 deletions(-)

diff --git a/datafusion/core/src/execution/context.rs b/datafusion/core/src/execution/context.rs
index 4ce48a064dbc..ce38892511f8 100644
--- a/datafusion/core/src/execution/context.rs
+++ b/datafusion/core/src/execution/context.rs
@@ -384,44 +384,28 @@ impl SessionContext {
             LogicalPlan::DropTable(DropTable {
                 name, if_exists, ..
             }) => {
-                let tp = self.get_table_provider(name.as_str());
-                // if table exists and is a table, deregister it
-                if let Some(table_provider) = tp {
-                    if table_provider.table_type() == TableType::Base {
-                        self.deregister_table(name.as_str())?;
-                        return self.return_empty_dataframe();
-                    }
-                }
-                // table doesn't exist or isn't a table, so either return quietly or show error
-                if if_exists {
-                    self.return_empty_dataframe()
-                } else {
-                    Err(DataFusionError::Execution(format!(
+                let result = self.find_and_deregister(name.as_str(), TableType::Base);
+                match (result, if_exists) {
+                    (Ok(true), _) => self.return_empty_dataframe(),
+                    (_, true) => self.return_empty_dataframe(),
+                    (_, _) => Err(DataFusionError::Execution(format!(
                         "Table {:?} doesn't exist.",
                         name
-                    )))
+                    ))),
                 }
             }
 
             LogicalPlan::DropView(DropView {
                 name, if_exists, ..
             }) => {
-                let tp = self.get_table_provider(name.as_str());
-                // if view exists and is a view, deregister it
-                if let Some(table_provider) = tp {
-                    if table_provider.table_type() == TableType::View {
-                        self.deregister_table(name.as_str())?;
-                        return self.return_empty_dataframe();
-                    }
-                }
-                // view doesn't exist or isn't a view, so either return quietly or show error
-                if if_exists {
-                    self.return_empty_dataframe()
-                } else {
-                    Err(DataFusionError::Execution(format!(
+                let result = self.find_and_deregister(name.as_str(), TableType::View);
+                match (result, if_exists) {
+                    (Ok(true), _) => self.return_empty_dataframe(),
+                    (_, true) => self.return_empty_dataframe(),
+                    (_, _) => Err(DataFusionError::Execution(format!(
                         "View {:?} doesn't exist.",
                         name
-                    )))
+                    ))),
                 }
             }
             LogicalPlan::CreateCatalogSchema(CreateCatalogSchema {
@@ -496,6 +480,26 @@ impl SessionContext {
         Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))
     }
 
+    fn find_and_deregister<'a>(
+        &self,
+        table_ref: impl Into<TableReference<'a>>,
+        table_type: TableType,
+    ) -> Result<bool> {
+        let table_ref = table_ref.into();
+        let table_provider = self
+            .state
+            .read()
+            .schema_for_ref(table_ref)?
+            .table(table_ref.table());
+
+        if let Some(table_provider) = table_provider {
+            if table_provider.table_type() == table_type {
+                self.deregister_table(table_ref)?;
+                return Ok(true);
+            }
+        }
+        Ok(false)
+    }
     /// Creates a logical plan.
     ///
     /// This function is intended for internal use and should not be called directly.
@@ -965,27 +969,6 @@ impl SessionContext {
     pub fn state(&self) -> SessionState {
         self.state.read().clone()
     }
-
-    /// Retrieves a TableProvider representing a table previously registered by calling the
-    /// register_table function.
-    ///
-    /// Returns none if no table has been registered with the provided reference.
-    fn get_table_provider<'a>(
-        &self,
-        table_ref: impl Into<TableReference<'a>>,
-    ) -> Option<Arc<dyn TableProvider>> {
-        let table_ref = table_ref.into();
-        let table = self
-            .state
-            .read()
-            .schema_for_ref(table_ref)
-            .ok()?
-            .table(table_ref.table());
-        match table {
-            Some(_) => table,
-            _ => None,
-        }
-    }
 }
 
 impl FunctionRegistry for SessionContext {

From 97cbb878a40f2500abf2bba17e32cbf95a2457e6 Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Fri, 26 Aug 2022 09:50:12 -0400
Subject: [PATCH 7/8] removed typo

---
 docs/source/user-guide/sql/ddl.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/source/user-guide/sql/ddl.md b/docs/source/user-guide/sql/ddl.md
index dc3f971171ac..8d24a8e4ce58 100644
--- a/docs/source/user-guide/sql/ddl.md
+++ b/docs/source/user-guide/sql/ddl.md
@@ -109,7 +109,7 @@ DROP TABLE IF EXISTS nonexistent_table;
 
 ## DROP VIEW
 
-Removes the view from DataFusion's catalog.add
+Removes the view from DataFusion's catalog.
 
 <pre>
 DROP VIEW [ IF EXISTS ] <b><i>view_name</i></b>;

From 7650489b1dd455c46126f269040be55adc35b257 Mon Sep 17 00:00:00 2001
From: kmitchener <kmitchener@gmail.com>
Date: Tue, 30 Aug 2022 20:31:53 -0400
Subject: [PATCH 8/8] fix bad resolve merge conflict

---
 datafusion/core/tests/sql/create_drop.rs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/datafusion/core/tests/sql/create_drop.rs b/datafusion/core/tests/sql/create_drop.rs
index 88773d65a4e5..820e570fac47 100644
--- a/datafusion/core/tests/sql/create_drop.rs
+++ b/datafusion/core/tests/sql/create_drop.rs
@@ -389,3 +389,4 @@ async fn create_csv_table_empty_file() -> Result<()> {
 /// Execute SQL and return results
 async fn plan_and_collect(ctx: &SessionContext, sql: &str) -> Result<Vec<RecordBatch>> {
     ctx.sql(sql).await?.collect().await
+}
