feat: materialized view declarations on local tables

A materialized view is a table whose contents are defined by a query over
one source table and maintained by refresh rather than by writes. The declaration half: create_materialized_view(name, source) resolves a
projected, filtered and limited definition against the source schema --
output types come from the DataFusion planner, never the caller -- and
commits an empty table carrying it as kind-tagged JSON in schema metadata.
The tag lets a kind added later read back as a view this version cannot
refresh rather than as a plain table.
Views open and list as ordinary tables.

Sources must have stable row ids, checked here because the property cannot
be enabled later: each view row records its source row in __source_row_id,
and that provenance survives compactions, updates and deletes only when row
ids are stable.

A view inherits the metadata describing its columns and none governing how a
table is written, so blob markers carry through while declarations its
always-nullable fields would contradict are stripped. Embedding
configuration is rewritten to the view's column names, and dropped where it
does not project both ends of a function.

rust/lancedb/src/connection.rs

@@ -41,7 +41,7 @@ use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
 
 mod create_table;
 
-fn merge_storage_options(
+pub(crate) fn merge_storage_options(
     store_params: &mut ObjectStoreParams,
     pairs: impl IntoIterator<Item = (String, String)>,
 ) {

rust/lancedb/src/database/listing.rs

@@ -819,6 +819,10 @@ impl ListingDatabase {
             .clone()
             .unwrap_or_default();
 
+        if let Some(store_params) = write_params.store_params.as_mut() {
+            strip_new_table_creation_keys(store_params);
+        }
+
         // 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
@@ -1288,8 +1292,196 @@ impl Database for ListingDatabase {
     }
 }
 
+/// Remove the `new_table_*` creation keys from a request's store options:
+/// they are creation configuration, not store credentials, and left in place
+/// they fork a fresh store connection for the request. An accessor carrying
+/// none of them is left untouched -- rebuilding a provider-backed accessor
+/// around a cached map would stop its credentials being fetched.
+fn strip_new_table_creation_keys(store_params: &mut ObjectStoreParams) {
+    let mut options = store_params.storage_options().cloned().unwrap_or_default();
+    let mut removed = false;
+    for key in [
+        OPT_NEW_TABLE_STORAGE_VERSION,
+        OPT_NEW_TABLE_V2_MANIFEST_PATHS,
+        OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS,
+    ] {
+        removed |= options.remove(key).is_some();
+    }
+    if !removed {
+        return;
+    }
+    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,
+        (true, Some(provider)) => Some(Arc::new(StorageOptionsAccessor::with_provider(provider))),
+        (false, Some(provider)) => Some(Arc::new(
+            StorageOptionsAccessor::with_initial_and_provider(options, provider),
+        )),
+        (false, None) => Some(Arc::new(StorageOptionsAccessor::with_static_options(
+            options,
+        ))),
+    };
+}
+
 #[cfg(test)]
 mod tests {
+    mod strip_new_table_creation_keys {
+        use super::super::*;
+
+        #[derive(Debug)]
+        struct EmptyProvider;
+
+        #[async_trait::async_trait]
+        impl StorageOptionsProvider for EmptyProvider {
+            async fn fetch_storage_options(
+                &self,
+            ) -> lance_core::Result<Option<HashMap<String, String>>> {
+                Ok(Some(HashMap::new()))
+            }
+
+            fn provider_id(&self) -> String {
+                "empty-test-provider".into()
+            }
+        }
+
+        fn params_with_static(options: &[(&str, &str)]) -> ObjectStoreParams {
+            ObjectStoreParams {
+                storage_options_accessor: Some(Arc::new(
+                    StorageOptionsAccessor::with_static_options(
+                        options
+                            .iter()
+                            .map(|(k, v)| (k.to_string(), v.to_string()))
+                            .collect(),
+                    ),
+                )),
+                ..Default::default()
+            }
+        }
+
+        #[test]
+        fn creation_keys_are_removed_and_store_keys_kept() {
+            let mut params = params_with_static(&[
+                ("region", "us-west-2"),
+                (OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true"),
+            ]);
+            strip_new_table_creation_keys(&mut params);
+            let options = params.storage_options().cloned().unwrap();
+            assert_eq!(options.get("region").map(String::as_str), Some("us-west-2"));
+            assert!(!options.contains_key(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS));
+        }
+
+        /// A request carrying only creation keys keeps the connection's
+        /// store: no accessor survives to fork a new one.
+        #[test]
+        fn only_creation_keys_leaves_no_accessor() {
+            let mut params = params_with_static(&[(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true")]);
+            strip_new_table_creation_keys(&mut params);
+            assert!(params.storage_options_accessor.is_none());
+        }
+
+        /// An accessor carrying no creation keys is not rebuilt: rebuilding a
+        /// provider-backed accessor around a cached map would stop its
+        /// credentials being fetched.
+        #[test]
+        fn untouched_provider_accessor_is_preserved() {
+            let accessor = Arc::new(StorageOptionsAccessor::with_provider(Arc::new(
+                EmptyProvider,
+            )));
+            let mut params = ObjectStoreParams {
+                storage_options_accessor: Some(accessor.clone()),
+                ..Default::default()
+            };
+            strip_new_table_creation_keys(&mut params);
+            assert!(Arc::ptr_eq(
+                params.storage_options_accessor.as_ref().unwrap(),
+                &accessor
+            ));
+        }
+
+        /// Stripping every static option from a provider-backed accessor must
+        /// yield a first-fetch accessor, not one caching an empty map.
+        #[test]
+        fn emptied_provider_accessor_still_fetches() {
+            let mut params = ObjectStoreParams {
+                storage_options_accessor: Some(Arc::new(
+                    StorageOptionsAccessor::with_initial_and_provider(
+                        HashMap::from([(
+                            OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS.to_string(),
+                            "true".to_string(),
+                        )]),
+                        Arc::new(EmptyProvider),
+                    ),
+                )),
+                ..Default::default()
+            };
+            strip_new_table_creation_keys(&mut params);
+            let accessor = params.storage_options_accessor.unwrap();
+            assert!(accessor.has_provider());
+            assert!(accessor.initial_storage_options().is_none());
+        }
+
+        /// Residual static options and the provider must both survive the
+        /// strip; losing the provider here would go unnoticed by the other
+        /// cases.
+        #[test]
+        fn mixed_static_and_provider_accessor_keeps_both() {
+            let mut params = ObjectStoreParams {
+                storage_options_accessor: Some(Arc::new(
+                    StorageOptionsAccessor::with_initial_and_provider(
+                        HashMap::from([
+                            ("region".to_string(), "us-west-2".to_string()),
+                            (
+                                OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS.to_string(),
+                                "true".to_string(),
+                            ),
+                        ]),
+                        Arc::new(EmptyProvider),
+                    ),
+                )),
+                ..Default::default()
+            };
+            strip_new_table_creation_keys(&mut params);
+            let accessor = params.storage_options_accessor.unwrap();
+            assert!(accessor.has_provider());
+            let residual = accessor.initial_storage_options().unwrap();
+            assert_eq!(
+                residual.get("region").map(String::as_str),
+                Some("us-west-2")
+            );
+            assert!(!residual.contains_key(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS));
+        }
+
+        /// End to end: a create carrying only a creation key stays on the
+        /// connection's store and the option takes effect.
+        #[tokio::test]
+        async fn creation_key_only_create_stays_on_the_connection_store() {
+            use arrow_array::record_batch;
+
+            let conn = crate::connect("memory://").execute().await.unwrap();
+            let batch = record_batch!(("x", Int32, [1])).unwrap();
+            conn.create_table("t", batch)
+                .storage_option(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true")
+                .execute()
+                .await
+                .unwrap();
+
+            let table = conn.open_table("t").execute().await.unwrap();
+            let stable = table
+                .as_native()
+                .unwrap()
+                .dataset
+                .get()
+                .await
+                .unwrap()
+                .manifest
+                .uses_stable_row_ids();
+            assert!(stable);
+        }
+    }
+
     use super::*;
     use crate::Table;
     use crate::arrow::{SendableRecordBatchStream, SimpleRecordBatchStream};

rust/lancedb/src/error.rs

@@ -77,6 +77,8 @@ pub enum Error {
     ColumnAlreadyExists { name: String },
     #[snafu(display("Column '{name}' is not a computed column"))]
     NotAComputedColumn { name: String },
+    #[snafu(display("Table '{name}' is not a materialized view"))]
+    NotAMaterializedView { name: String },
     #[snafu(display("Invalid expression for column '{column}': {message}"))]
     InvalidExpression { column: String, message: String },
 

rust/lancedb/src/lib.rs

@@ -186,6 +186,7 @@ pub mod index;
 pub mod io;
 pub mod ipc;
 pub mod job;
+pub mod materialized_view;
 #[cfg(feature = "metrics-otel")]
 pub mod metrics_otel;
 #[cfg(feature = "polars")]
@@ -210,6 +211,7 @@ pub use function::FunctionVersion;
 pub use job::Job;
 use lance_index::vector::ApproxMode as LanceApproxMode;
 use lance_linalg::distance::DistanceType as LanceDistanceType;
+pub use materialized_view::{MaterializedView, MaterializedViewDefinition};
 /// Re-export of the [`metrics`](https://docs.rs/metrics) crate facade. Enable
 /// the `metrics` feature to publish LanceDB's internal metrics; install any
 /// `metrics`-compatible recorder to collect them. See also [`metrics_otel`] for

rust/lancedb/src/remote/table.rs

@@ -10359,6 +10359,20 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_materialized_view_refused_without_a_request() {
+        // Materialized views are local-only. The table-level entry the
+        // bindings use must refuse a remote table before reading its schema,
+        // so the panicking handler is the assertion.
+        let table = Table::new_with_handler("my_table", |request| -> http::Response<String> {
+            panic!("unexpected request: {}", request.url().path())
+        });
+        let err = crate::MaterializedView::from_table(table)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::NotSupported { .. }), "got {err:?}");
+    }
+
     #[tokio::test]
     async fn test_create_branch_empty_name_rejected_client_side() {
         use lance::dataset::refs::Ref;

rust/lancedb/src/table.rs

@@ -1059,6 +1059,11 @@ impl Table {
         self.database.as_ref().unwrap()
     }
 
+    /// The database this handle was opened through, when it was.
+    pub fn database_opt(&self) -> Option<&Arc<dyn Database>> {
+        self.database.as_ref()
+    }
+
     pub fn embedding_registry(&self) -> &Arc<dyn EmbeddingRegistry> {
         &self.embedding_registry
     }

rust/lancedb/src/table/refresh.rs

@@ -193,7 +193,7 @@ fn declared_expression(dataset: &Dataset, column: &str) -> Result<String> {
 ///
 /// Lance's dialect delimits with backticks, so a double-quoted name would
 /// parse as a string literal rather than a column.
-fn quote_identifier(name: &str) -> String {
+pub(crate) fn quote_identifier(name: &str) -> String {
     format!("`{}`", name.replace('`', "``"))
 }