feat: allow COPY to be disabled independently (#25824)

## Which issue does this PR close?

- Closes #25823.

## Rationale for this change

Applications embedding DataFusion may want to allow DML operations such
as INSERT into registered table providers while disallowing COPY, which
writes query results to a file or object store. SQLOptions currently
controls COPY through allow_dml, so COPY cannot be disabled without also
disabling other DML.

## What changes are included in this PR?

- Add SQLOptions::with_allow_copy to control COPY independently,
defaulting to true.
- Keep with_allow_dml(false) behavior unchanged so it continues to
reject COPY.
- Add an integration test showing that disabling COPY does not disable
INSERT.

## What is the testing strategy for this PR?

- cargo fmt --all -- --check
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test -p datafusion --test core_integration --
sql::sql_api::copy_can_be_disabled_independently_from_dml --exact
- cargo test -p datafusion --test core_integration --
sql::sql_api::unsupported_copy_returns_error --exact

## Are there any user-facing changes?

Yes. This adds the public SQLOptions::with_allow_copy builder method. It
is backwards compatible because COPY remains enabled by default and
with_allow_dml(false) continues to reject COPY.

Author: shyjsarah

datafusion/core/src/execution/context/mod.rs

@@ -2358,6 +2358,8 @@ pub struct SQLOptions {
     allow_ddl: bool,
     /// See [`Self::with_allow_dml`]
     allow_dml: bool,
+    /// See [`Self::with_allow_copy`]
+    allow_copy: bool,
     /// See [`Self::with_allow_statements`]
     allow_statements: bool,
 }
@@ -2367,6 +2369,7 @@ impl Default for SQLOptions {
         Self {
             allow_ddl: true,
             allow_dml: true,
+            allow_copy: true,
             allow_statements: true,
         }
     }
@@ -2390,6 +2393,15 @@ impl SQLOptions {
         self
     }
 
+    /// Should `COPY` commands be run? Defaults to `true`.
+    ///
+    /// `COPY` commands are also disabled when DML is disabled with
+    /// [`Self::with_allow_dml`].
+    pub fn with_allow_copy(mut self, allow: bool) -> Self {
+        self.allow_copy = allow;
+        self
+    }
+
     /// Should Statements such as (e.g. `SET VARIABLE and `BEGIN TRANSACTION` ...`) be run?. Defaults to `true`
     pub fn with_allow_statements(mut self, allow: bool) -> Self {
         self.allow_statements = allow;
@@ -2424,6 +2436,9 @@ impl<'n> TreeNodeVisitor<'n> for BadPlanVisitor<'_> {
             LogicalPlan::Dml(dml) if !self.options.allow_dml => {
                 plan_err!("DML not supported: {}", dml.op)
             }
+            LogicalPlan::Copy(_) if !self.options.allow_copy => {
+                plan_err!("COPY not supported")
+            }
             LogicalPlan::Copy(_) if !self.options.allow_dml => {
                 plan_err!("DML not supported: COPY")
             }

datafusion/core/tests/sql/sql_api.rs

@@ -17,7 +17,7 @@
 
 use datafusion::prelude::*;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
-use datafusion_common::{TableReference, assert_contains};
+use datafusion_common::{TableReference, assert_batches_eq, assert_contains};
 use datafusion_expr::dml::MergeIntoOp;
 use datafusion_expr::{Expr, LogicalPlan, WriteOp};
 
@@ -116,6 +116,45 @@ async fn unsupported_copy_returns_error() {
     ctx.sql_with_options(&sql, options).await.unwrap();
 }
 
+#[tokio::test]
+async fn copy_can_be_disabled_independently_from_dml() {
+    let tmpdir = TempDir::new().unwrap();
+    let tmpfile = tmpdir.path().join("foo.parquet");
+
+    let ctx = SessionContext::new();
+    ctx.sql("CREATE TABLE test (x int)").await.unwrap();
+
+    let options = SQLOptions::new().with_allow_copy(false);
+    let sql = format!(
+        "COPY (values(1)) TO '{}' STORED AS parquet",
+        tmpfile.to_string_lossy()
+    );
+    let df = ctx.sql_with_options(&sql, options).await;
+    assert_eq!(
+        df.unwrap_err().strip_backtrace(),
+        "Error during planning: COPY not supported"
+    );
+
+    ctx.sql_with_options("INSERT INTO test VALUES (1)", options)
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    let results = ctx
+        .sql_with_options("SELECT x FROM test", options)
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+    assert_batches_eq!(["+---+", "| x |", "+---+", "| 1 |", "+---+"], &results);
+
+    let options = options.with_allow_copy(true);
+    ctx.sql_with_options(&sql, options).await.unwrap();
+}
+
 #[tokio::test]
 async fn unsupported_statement_returns_error() {
     let ctx = SessionContext::new();