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. This adds 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 at every depth while declarations its always-nullable fields would contradict are stripped. Embedding configuration and column definitions are rewritten to the names the view gives those columns, and dropped where it does not project both ends of a function.

Refresh lands in the next layer of this stack; a declared view is inert but introspectable.

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

@@ -185,6 +185,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")]
@@ -208,6 +209,7 @@ pub use error::{Error, JobFailure, Result};
 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

@@ -10156,6 +10156,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

@@ -1058,6 +1058,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

@@ -182,7 +182,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('`', "``"))
 }