diff --git a/rust/lancedb/src/materialized_view.rs b/rust/lancedb/src/materialized_view.rs index a5f2e0fd5..bd77b3340 100644 --- a/rust/lancedb/src/materialized_view.rs +++ b/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::>(); + 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::>() .join(", ") }; - let source = self - .source_namespace - .iter() - .chain(std::iter::once(&self.source)) - .map(|part| quote(part)) - .collect::>() - .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(); diff --git a/rust/lancedb/src/materialized_view/grouped.rs b/rust/lancedb/src/materialized_view/grouped.rs index c52686bfd..3fb84a814 100644 --- a/rust/lancedb/src/materialized_view/grouped.rs +++ b/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; diff --git a/rust/lancedb/src/materialized_view/refresh.rs b/rust/lancedb/src/materialized_view/refresh.rs index 7f0360a0c..ca90f3214 100644 --- a/rust/lancedb/src/materialized_view/refresh.rs +++ b/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 { diff --git a/rust/lancedb/src/table/datafusion/udtf/vector_query.rs b/rust/lancedb/src/table/datafusion/udtf/vector_query.rs index d2cc90d27..000addbe1 100644 --- a/rust/lancedb/src/table/datafusion/udtf/vector_query.rs +++ b/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;