fix: address materialized view review findings

rust/lancedb/src/materialized_view.rs

@@ -1450,6 +1450,25 @@ pub struct CreateMaterializedViewBuilder {
     with_no_data: bool,
 }
 
+fn source_sql_identifier(
+    namespace_path: &[String],
+    source: &str,
+    namespace_in_schema_position: bool,
+) -> String {
+    fn quote(name: &str) -> String {
+        format!("\"{}\"", name.replace('"', "\"\""))
+    }
+
+    let namespace = if namespace_in_schema_position && !namespace_path.is_empty() {
+        vec![namespace_path.join("$")]
+    } else {
+        namespace_path.to_vec()
+    };
+    let mut parts = namespace.iter().map(|part| quote(part)).collect::<Vec<_>>();
+    parts.push(quote(source));
+    parts.join(".")
+}
+
 impl CreateMaterializedViewBuilder {
     pub(crate) fn new(connection: Connection, name: String, source: String) -> Self {
         Self {
@@ -1524,13 +1543,11 @@ impl CreateMaterializedViewBuilder {
                 .collect::<Vec<_>>()
                 .join(", ")
         };
-        let source = self
-            .source_namespace
-            .iter()
-            .chain(std::iter::once(&self.source))
-            .map(|part| quote(part))
-            .collect::<Vec<_>>()
-            .join(".");
+        let source = source_sql_identifier(
+            &self.source_namespace,
+            &self.source,
+            self.connection.uri().starts_with("db://"),
+        );
         let mut query = format!("SELECT {projection} FROM {source}");
         if let Some(filter) = &self.filter {
             query.push_str(" WHERE ");
@@ -1807,6 +1824,19 @@ mod tests {
     use super::*;
     use crate::connect;
 
+    #[test]
+    fn remote_source_namespace_uses_one_sql_schema_position() {
+        let namespace = vec!["analytics".to_string(), "daily".to_string()];
+        assert_eq!(
+            source_sql_identifier(&namespace, "source", true),
+            "\"analytics$daily\".\"source\""
+        );
+        assert_eq!(
+            source_sql_identifier(&namespace, "source", false),
+            "\"analytics\".\"daily\".\"source\""
+        );
+    }
+
     #[tokio::test]
     async fn refresh_always_rebuilds_without_source_row_ids() {
         let conn = connect("memory://").execute().await.unwrap();

rust/lancedb/src/materialized_view/grouped.rs

@@ -12,7 +12,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use arrow_array::cast::AsArray;
 use arrow_array::{Array, FixedSizeListArray, UInt32Array};
 use arrow_buffer::NullBuffer;
-use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef};
+use arrow_schema::{DataType, Schema as ArrowSchema, SchemaRef};
 use datafusion::catalog::default_table_source::provider_as_source;
 use datafusion::datasource::empty::EmptyTable;
 use datafusion::execution::FunctionRegistry;

rust/lancedb/src/materialized_view/refresh.rs

@@ -27,8 +27,8 @@ use serde::{Deserialize, Serialize};
 
 use super::{MaterializedViewDefinition, StoredDefinition};
 use crate::database::OpenTableRequest;
-use crate::table::Table;
 use crate::table::computed_columns::{computed_column_from_field, ensure_declarations_are_planned};
+use crate::table::{NativeTableExt, Table};
 use crate::{Error, Result};
 
 /// How a refresh brought the view up to date.
@@ -351,7 +351,7 @@ async fn compute_stream(
     Ok(Box::pin(RecordBatchStreamAdapter::new(schema, mapped)))
 }
 
-fn to_view_batch(
+pub(super) fn to_view_batch(
     batch: &RecordBatch,
     output_schema: &SchemaRef,
 ) -> datafusion::common::Result<RecordBatch> {

rust/lancedb/src/table/datafusion/udtf/vector_query.rs

@@ -25,11 +25,11 @@ use datafusion_physical_plan::{
     projection::ProjectionExec,
     stream::RecordBatchStreamAdapter,
 };
-use futures::{StreamExt, TryStreamExt, stream};
+use futures::{TryStreamExt, stream};
 use lance::dataset::scanner::{RowAddrMask, RowAddrTreeMap};
 use lance::{Dataset, index::DatasetIndexInternalExt};
 use lance_core::datatypes::BlobHandling;
-use lance_datafusion::exec::{OneShotExec, SessionContextExt};
+use lance_datafusion::exec::SessionContextExt;
 use lance_index::metrics::NoOpMetricsCollector;
 use roaring::RoaringTreemap;