Pin native schema modifiers and refuse lossy planning

Cargo.toml

@@ -196,7 +196,7 @@ regex = "1.12"
 rstest = "0.26.1"
 serde_json = "1"
 sha2 = "^0.11.0"
-sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "7fdc000a3309ef6ef79dc3c863aca0951ae93edb", default-features = false, features = [
+sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "874cfbb81d2a34e6ee101d3a7ecd1ae6b60000c6", default-features = false, features = [
     "std",
     "visitor",
 ] }

datafusion/sql/src/statement.rs

@@ -779,14 +779,24 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             Statement::CreateSchema {
                 schema_name,
                 if_not_exists,
+                or_replace,
+                transient,
+                with_managed_access,
                 ..
-            } => Ok(LogicalPlan::Ddl(DdlStatement::CreateCatalogSchema(
-                CreateCatalogSchema {
-                    schema_name: get_schema_name(&schema_name, self.ident_normalizer),
-                    if_not_exists,
-                    schema: Arc::new(DFSchema::empty()),
-                },
-            ))),
+            } => {
+                if or_replace || transient || with_managed_access {
+                    return not_impl_err!(
+                        "CREATE SCHEMA OR REPLACE, TRANSIENT and WITH MANAGED ACCESS are not supported"
+                    );
+                }
+                Ok(LogicalPlan::Ddl(DdlStatement::CreateCatalogSchema(
+                    CreateCatalogSchema {
+                        schema_name: get_schema_name(&schema_name, self.ident_normalizer),
+                        if_not_exists,
+                        schema: Arc::new(DFSchema::empty()),
+                    },
+                )))
+            }
             Statement::CreateDatabase {
                 db_name,
                 if_not_exists,

datafusion/sql/tests/sql_integration.rs

@@ -3933,6 +3933,31 @@ fn create_schema_with_quoted_name() {
     );
 }
 
+#[test]
+fn create_schema_modifiers_are_not_silently_discarded() {
+    for sql in [
+        "CREATE OR REPLACE SCHEMA s",
+        "CREATE TRANSIENT SCHEMA s",
+        "CREATE SCHEMA s WITH MANAGED ACCESS",
+        "CREATE OR REPLACE TRANSIENT SCHEMA s WITH MANAGED ACCESS",
+        "CREATE TRANSIENT SCHEMA IF NOT EXISTS s",
+    ] {
+        let error = logical_plan_with_dialect(sql, &SnowflakeDialect {})
+            .expect_err("unsupported schema modifiers must not become plain CREATE");
+        assert_contains!(
+            error.to_string(),
+            "CREATE SCHEMA OR REPLACE, TRANSIENT and WITH MANAGED ACCESS are not supported"
+        );
+    }
+    let plan =
+        logical_plan_with_dialect("CREATE SCHEMA IF NOT EXISTS s", &SnowflakeDialect {})
+            .expect("ordinary conditional CREATE remains supported");
+    let LogicalPlan::Ddl(DdlStatement::CreateCatalogSchema(create)) = plan else {
+        panic!("expected catalog schema plan");
+    };
+    assert!(create.if_not_exists);
+}
+
 #[test]
 fn create_schema_with_quoted_unnormalized_name() {
     let sql = "CREATE SCHEMA \"Foo\"";