fix: canonicalize materialized view filters

rust/lancedb/src/materialized_view.rs

@@ -170,6 +170,15 @@ pub(crate) fn plan(
     filter: Option<&str>,
     limit: Option<u64>,
 ) -> Result<(MaterializedViewDefinition, Vec<ArrowField>, Lineage)> {
+    let filter = filter
+        .map(crate::expr::canonicalize_sql_predicate)
+        .transpose()
+        .map_err(|err| match err {
+            Error::InvalidInput { message } => Error::InvalidInput {
+                message: format!("invalid view filter: {message}"),
+            },
+            err => err,
+        })?;
     let projections: Vec<(String, String)> = if projections.is_empty() {
         source_schema
             .fields()
@@ -274,7 +283,7 @@ pub(crate) fn plan(
         declared.push(output);
     }
 
-    if let Some(filter) = filter {
+    if let Some(filter) = filter.as_deref() {
         let expr = planner
             .parse_filter(filter)
             .map_err(|e| Error::InvalidInput {
@@ -314,7 +323,7 @@ pub(crate) fn plan(
             .into_iter()
             .map(|(output, expression)| ViewProjection { output, expression })
             .collect(),
-        filter: filter.map(String::from),
+        filter,
         limit,
         inputs,
     };

rust/lancedb/src/materialized_view/refresh.rs

@@ -1585,6 +1585,40 @@ mod tests {
         assert_eq!(read(view.table(), "x").await, vec![20, 40]);
     }
 
+    #[tokio::test]
+    async fn test_mixed_case_filter_is_canonicalized_for_lineage_and_refresh() {
+        let conn = connect("memory://").execute().await.unwrap();
+        let batch = record_batch!(
+            ("id", Int32, [1, 2, 3]),
+            ("PartyAbbrev", Utf8, ["D", "R", "D"])
+        )
+        .unwrap();
+        conn.create_table("src", batch)
+            .write_options(crate::materialized_view::tests::stable_row_ids())
+            .execute()
+            .await
+            .unwrap();
+        conn.create_materialized_view("democrats", "src")
+            .select([("id", "id")])
+            .only_if(r#""PartyAbbrev" = 'D'"#)
+            .execute()
+            .await
+            .unwrap();
+
+        // Reopen from schema metadata so these assertions cover the stored
+        // predicate and lineage, not only the declaration-time handle.
+        let view = conn.open_materialized_view("democrats").await.unwrap();
+        assert_eq!(
+            view.definition().filter.as_deref(),
+            Some("`PartyAbbrev` = 'D'")
+        );
+        assert_eq!(view.definition().inputs, ["PartyAbbrev", "id"]);
+
+        let result = view.refresh().execute().await.unwrap();
+        assert_eq!(result.rows_written, 2);
+        assert_eq!(read(view.table(), "id").await, vec![1, 3]);
+    }
+
     #[tokio::test]
     async fn test_append_refreshes_incrementally() {
         let (_conn, source, view) = refreshed_doubled(vec![1, 2]).await;