diff --git a/rust/lancedb/src/database/listing.rs b/rust/lancedb/src/database/listing.rs
index 0ab3614e7..300112339 100644
--- a/rust/lancedb/src/database/listing.rs
+++ b/rust/lancedb/src/database/listing.rs
@@ -819,6 +819,33 @@ impl ListingDatabase {
             .clone()
             .unwrap_or_default();
 
+        // The new-table keys are creation configuration, not store options:
+        // strip them so a request carrying only them keeps the connection's
+        // store instead of forking a new one.
+        if let Some(store_params) = write_params.store_params.as_mut() {
+            let mut options = store_params.storage_options().cloned().unwrap_or_default();
+            for key in [
+                OPT_NEW_TABLE_STORAGE_VERSION,
+                OPT_NEW_TABLE_V2_MANIFEST_PATHS,
+                OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS,
+            ] {
+                options.remove(key);
+            }
+            let provider = store_params
+                .storage_options_accessor
+                .as_ref()
+                .and_then(|accessor| accessor.provider().cloned());
+            store_params.storage_options_accessor = match (options.is_empty(), provider) {
+                (true, None) => None,
+                (_, Some(provider)) => Some(Arc::new(
+                    StorageOptionsAccessor::with_initial_and_provider(options, provider),
+                )),
+                (false, None) => Some(Arc::new(StorageOptionsAccessor::with_static_options(
+                    options,
+                ))),
+            };
+        }
+
         // Only modify the storage options if we actually have something to
         // inherit. There is a difference between storage_options=None and
         // storage_options=Some({}). Using storage_options=None will cause the
diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs
index 6f2a8157a..7d1595ada 100644
--- a/rust/lancedb/src/materialized_view.rs
+++ b/rust/lancedb/src/materialized_view.rs
@@ -28,6 +28,7 @@ use lance_datafusion::planner::Planner;
 use serde::{Deserialize, Serialize};
 
 use crate::connection::Connection;
+use crate::database::listing::OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS;
 use crate::table::refresh::quote_identifier;
 use crate::table::{Table, WriteOptions};
 use crate::{Error, Result};
@@ -463,16 +464,14 @@ impl CreateMaterializedViewBuilder {
             definition_to_metadata(&definition)?,
         )]);
         let schema = Arc::new(ArrowSchema::new_with_metadata(fields, metadata));
-        // Stable row ids so the view can itself source another view.
+        // Stable row ids so the view can itself source another view. The
+        // request-level option outranks any connection-level override; the
+        // manifest is still verified because another Database impl may
+        // ignore the option, but nothing is rolled back on failure.
         let table = self
             .connection
             .create_empty_table(&self.name, schema)
-            .write_options(WriteOptions {
-                lance_write_params: Some(lance::dataset::WriteParams {
-                    enable_stable_row_ids: true,
-                    ..Default::default()
-                }),
-            })
+            .storage_option(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true")
             .execute()
             .await?;
         let stable = match table.as_native() {
@@ -480,13 +479,11 @@ impl CreateMaterializedViewBuilder {
             None => false,
         };
         if !stable {
-            let _ = self.connection.drop_table(&self.name, &[]).await;
-            return Err(Error::InvalidInput {
+            return Err(Error::Runtime {
                 message: format!(
-                    "view '{}' would be created without stable row ids: the \
-                     connection's new_table_enable_stable_row_ids setting \
-                     overrides view creation; remove it or use a connection \
-                     without it",
+                    "view '{}' was created without stable row ids: the database \
+                     ignored the creation option; the table remains and is not \
+                     usable as a materialized view",
                     self.name
                 ),
             });
@@ -882,42 +879,41 @@ mod tests {
         assert_eq!(definition.filter.as_deref(), Some("age >= 18"));
     }
 
-    /// A connection override that would create the view without stable row
-    /// ids fails loudly instead of committing an unchainable view.
+    /// The creation option outranks a connection configured to create
+    /// unstable tables: the view still gets stable row ids, on the same
+    /// store (no fork -- the table must be reachable through the
+    /// connection afterwards).
     #[tokio::test]
-    async fn test_conflicting_connection_override_is_refused() {
-        let dir = tempfile::tempdir().unwrap();
-        let uri = dir.path().to_str().unwrap();
-        let stable_conn = connect(uri).execute().await.unwrap();
-        let batch = record_batch!(("x", Int32, [1])).unwrap();
-        stable_conn
-            .create_table("src", batch)
-            .write_options(stable_row_ids())
-            .execute()
-            .await
-            .unwrap();
-
-        let unstable_conn = connect(uri)
+    async fn test_view_is_stable_despite_connection_override() {
+        let conn = connect("memory://")
             .storage_options([("new_table_enable_stable_row_ids", "false")])
             .execute()
             .await
             .unwrap();
-        let err = unstable_conn
+        let batch = record_batch!(("x", Int32, [1])).unwrap();
+        conn.create_table("src", batch)
+            .storage_option("new_table_enable_stable_row_ids", "true")
+            .execute()
+            .await
+            .unwrap();
+
+        let view = conn
             .create_materialized_view("v", "src")
             .execute()
             .await
-            .unwrap_err();
-        assert!(
-            matches!(err, Error::InvalidInput { message } if message.contains("stable row ids"))
-        );
-        assert!(
-            !unstable_conn
-                .table_names()
-                .execute()
-                .await
-                .unwrap()
-                .contains(&"v".to_string())
-        );
+            .unwrap();
+        let stable = view
+            .table()
+            .as_native()
+            .unwrap()
+            .dataset
+            .get()
+            .await
+            .unwrap()
+            .manifest
+            .uses_stable_row_ids();
+        assert!(stable);
+        conn.open_materialized_view("v").await.unwrap();
     }
 
     /// A committed filter has to be usable as a predicate.