fix: skip insert when creating table that exists (#2901)

Co-authored-by: Tal Gluck <talagluck@gmail.com>

crates/sqlexec/src/planner/physical_plan/create_table.rs

@@ -188,37 +188,45 @@ impl CreateTableExec {
             }
         };
 
-        let table = storage.create_table(ent, save_mode).await.map_err(|e| {
-            DataFusionError::Execution(format!("failed to create table in storage: {e}"))
-        })?;
+        let table_existed = storage
+            .table_exists(ent)
+            .await
+            .map_err(|e| DataFusionError::External(Box::new(e)))?;
 
-        let insert_res = match (source, or_replace) {
-            (Some(input), overwrite) => insert(&table, input, overwrite, context).await,
+        if !table_existed || !if_not_exists {
+            let table = storage.create_table(ent, save_mode).await.map_err(|e| {
+                DataFusionError::Execution(format!("failed to create table in storage: {e}"))
+            })?;
 
-            // if it's a 'replace' and there is no insert, we overwrite with an empty table
-            (None, true) => {
-                let input = Arc::new(EmptyExec::new(TableProvider::schema(&table)));
-                insert(&table, input, true, context).await
+            let insert_res = match (source, or_replace) {
+                (Some(input), overwrite) => insert(&table, input, overwrite, context).await,
+
+                // if it's a 'replace' and there is no insert, we overwrite with an empty table
+                (None, true) => {
+                    let input = Arc::new(EmptyExec::new(TableProvider::schema(&table)));
+                    insert(&table, input, true, context).await
+                }
+                (None, false) => Ok(()),
+            };
+
+            if let Err(e) = insert_res {
+                storage.delete_table(ent).await.map_err(|e| {
+                    DataFusionError::Execution(format!("failed to clean up table: {e}"))
+                })?;
+                return Err(e);
             }
-            (None, false) => Ok(()),
-        };
 
-        if let Err(e) = insert_res {
-            storage.delete_table(ent).await.map_err(|e| {
-                DataFusionError::Execution(format!("failed to clean up table: {e}"))
-            })?;
-            return Err(e);
-        }
+            mutator
+                .commit_state(catalog_version, state.as_ref().clone())
+                .await
+                .map_err(|e| {
+                    DataFusionError::Execution(format!("failed to commit catalog state: {e}"))
+                })?;
 
-        mutator
-            .commit_state(catalog_version, state.as_ref().clone())
-            .await
-            .map_err(|e| {
-                DataFusionError::Execution(format!("failed to commit catalog state: {e}"))
-            })?;
-        debug!(loc = %table.storage_location(), "native table created");
+            debug!(loc = %table.storage_location(), "native table created");
 
-        // TODO: Add storage tracking job.
+            // TODO: Add storage tracking job.
+        }
 
         Ok(new_operation_batch("create_table"))
     }

testdata/sqllogictests/create_table.slt

@@ -135,3 +135,19 @@ create table test as select vector, point as "Point" from lance_scan('./testdata
 
 statement ok
 select * from test;
+
+statement ok
+create table if not exists foo as select 1;
+
+query I
+select * from foo;
+----
+1
+
+statement ok
+create table if not exists foo as select 'a'
+
+query I
+select * from foo;
+----
+1