diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs
index 4cca4c114a91..7717f75d16b8 100644
--- a/datafusion/sql/src/statement.rs
+++ b/datafusion/sql/src/statement.rs
@@ -850,7 +850,7 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                     return plan_err!("Unsupported Value in COPY statement {}", value);
                 }
             };
-            if !(&key.contains('.')) {
+            if !(key.contains('.') || key == "format") {
                 // If config does not belong to any namespace, assume it is
                 // a format option and apply the format prefix for backwards
                 // compatibility.
@@ -866,12 +866,16 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             FileType::from_str(&file_type).map_err(|_| {
                 DataFusionError::Configuration(format!("Unknown FileType {}", file_type))
             })?
+        } else if let Some(format) = options.remove("format") {
+            // try to infer file format from the "format" key in options
+            FileType::from_str(&format)
+                .map_err(|e| DataFusionError::Configuration(format!("{}", e)))?
         } else {
             let e = || {
                 DataFusionError::Configuration(
-                "Format not explicitly set and unable to get file extension! Use STORED AS to define file format."
-                    .to_string(),
-            )
+                    "Format not explicitly set and unable to get file extension! Use STORED AS to define file format."
+                        .to_string(),
+                )
             };
             // try to infer file format from file extension
             let extension: &str = &Path::new(&statement.target)
diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs
index 47638e58ff00..c738a2bd754f 100644
--- a/datafusion/sql/tests/sql_integration.rs
+++ b/datafusion/sql/tests/sql_integration.rs
@@ -442,6 +442,18 @@ CopyTo: format=csv output_url=output.csv options: ()
     quick_test(sql, plan);
 }
 
+#[test]
+fn plan_copy_stored_as_priority() {
+    let sql = "COPY (select * from (values (1))) to 'output/' STORED AS CSV OPTIONS (format json)";
+    let plan = r#"
+CopyTo: format=csv output_url=output/ options: (format json)
+  Projection: column1
+    Values: (Int64(1))
+    "#
+    .trim();
+    quick_test(sql, plan);
+}
+
 #[test]
 fn plan_insert() {
     let sql =
diff --git a/datafusion/sqllogictest/test_files/copy.slt b/datafusion/sqllogictest/test_files/copy.slt
index 75f1ccb07aac..fca892dfcdad 100644
--- a/datafusion/sqllogictest/test_files/copy.slt
+++ b/datafusion/sqllogictest/test_files/copy.slt
@@ -514,6 +514,44 @@ OPTIONS (
 );
 
 
+# Format Options Support with format in OPTIONS i.e. COPY { table_name | query } TO 'file_name' OPTIONS (format <format-name>, ...)
+
+query I
+COPY (select * from (values (1))) to 'test_files/scratch/copy/'
+OPTIONS (format parquet);
+----
+1
+
+query I
+COPY (select * from (values (1))) to 'test_files/scratch/copy/'
+OPTIONS (format parquet, compression 'zstd(10)');
+----
+1
+
+query I
+COPY (select * from (values (1))) to 'test_files/scratch/copy/'
+OPTIONS (format json, compression gzip);
+----
+1
+
+query I
+COPY (select * from (values (1))) to 'test_files/scratch/copy/'
+OPTIONS (
+    format csv,
+    has_header false,
+    compression xz,
+    datetime_format '%FT%H:%M:%S.%9f',
+    delimiter ';',
+    null_value 'NULLVAL'
+);
+----
+1
+
+query error DataFusion error: Invalid or Unsupported Configuration: This feature is not implemented: Unknown FileType: NOTVALIDFORMAT
+COPY (select * from (values (1))) to 'test_files/scratch/copy/'
+OPTIONS (format notvalidformat, compression 'zstd(5)');
+
+
 # Error cases:
 
 # Copy from table with options