feat(linux): per-table WHERE filter dialog with full operator set

Server-side filter for the Browse tab with per-column operators,
type-aware value coercion, and per-(connection, schema, table)
persistence. Reachable via a Filter button on the paginator action
bar and Ctrl+R shortcut.

Architecture (3 layers, 3 new files):

- crates/core/src/filter.rs: FilterSet / FilterRule / FilterOp /
  FilterValue / Combinator types + build_filter_where(driver_id,
  columns, set) -> Result<Option<(String, Vec<Value>)>, BuildFilterError>.
  Returns Ok(None) for empty rules so callers skip WHERE entirely.
  Identifiers via sql_dialect::quote_ident; placeholders via
  placeholder_for ($N for PG, ? for MySQL/SQLite). Type-aware
  coercion: "42" against an int column binds as Value::Int(42),
  not Value::Text. PG ILIKE; falls back to LIKE on MySQL/SQLite.
  29 unit tests cover every operator, all three drivers' placeholder
  dialects, type coercion (int/float/decimal/date/datetime/timestamptz/
  uuid/bool/json), error paths, and serde round-trip.

- crates/app/src/services/filter_settings.rs: per-(connection_id,
  schema, table) persistence in $XDG_CONFIG_HOME/tablepro/filter_settings.json.
  Atomic temp-file rename via existing config_io::atomic_write_json.
  Empty FilterSet save removes the entry so the file shrinks back.
  Schema is in the key (column_widths.json doesn't include schema —
  latent bug there, not fixed in this commit). 3 unit tests.

- crates/app/src/ui/filter_dialog.rs: AdwDialog with the rule editor.
  Match-combinator at the top (DropDown: All / Any), boxed-list of
  rules below, Add Rule footer row. Value widget shape adapts to
  operator (None for IsNull, Single for Eq/Lt, Pair for BETWEEN,
  List for IN). Column-type narrows the operator dropdown via per-
  type allowlist mirroring the core classifier. Bytes columns hidden
  from the column dropdown. Cancel / Clear all / Apply in the
  HeaderBar; default widget = Apply so Enter submits.

Connection trait: new `query_params(sql, params)` with a default
that delegates to `query` for empty params (legacy callers
unchanged) and errors otherwise. PG / MySQL / SQLite drivers
override with sqlx parametric streaming, mirroring the existing
execute_params shape. ReadOnlyConnection passes through (filter
SELECTs are still safe).

Browse plumbing:
- BrowseTab gains current_filter (FilterSet) restored from
  filter_settings on init.
- fetch_browse_page composes SELECT * FROM ... WHERE <built> ORDER
  BY ... LIMIT ... OFFSET ... and routes via query_params.
  No-WHERE + no-ORDER BY keeps the fast-path fetch_rows call so
  unfiltered usage doesn't pay the parametric overhead.
- fetch_browse_row_count applies the same WHERE so the "of N" total
  reflects filtered counts. Validation errors silently skip there
  (the page fetch on the same tick already toasts).
- FilterApplied input persists, resets offset to 0, refreshes
  chrome, fires both FetchPage and FetchRowCount.
- Filter button picks up `.accent` CSS class + count-suffixed
  tooltip when rules are active.

App:
- new AppMsg::ShowFilterDialog routes to on_show_filter_dialog
  which fetches the active tab's columns + current filter and
  presents the dialog with an on_apply callback that sends
  BrowseTabInput::FilterApplied.
- Ctrl+R bound at window scope; `win.open-filter` action; entry
  in the Browse-table section of the shortcuts window.

92 + 103 unit tests pass (29 new filter tests + 3 filter_settings
tests). Manual GUI walkthrough deferred to in-app verification.

Author: datlechin

linux/crates/app/src/services/filter_settings.rs

@@ -0,0 +1,167 @@
+//! Per-table filter persistence. Mirrors `column_widths.rs` but the
+//! key includes the schema (so `public.users` and `audit.users` don't
+//! collide on a multi-schema Postgres database) and the value is a
+//! `FilterSet` rather than a single integer.
+//!
+//! Save path is debounced via `relm4::spawn`; rapid Apply clicks land
+//! one disk write each, but they don't block the GTK main loop. An
+//! empty `FilterSet` removes the entry from the file so it shrinks
+//! back when the user clears their filter.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+use tablepro_core::FilterSet;
+use uuid::Uuid;
+
+use super::config_io::{atomic_write_json, xdg_config_path};
+
+const FILE: &str = "filter_settings.json";
+
+/// `FilterSet` keyed by `(connection_id, schema-or-empty, table)`.
+type Tables = HashMap<String, FilterSet>;
+type Schemas = HashMap<String, Tables>;
+type Connections = HashMap<String, Schemas>;
+
+static CACHE: Mutex<Option<Connections>> = Mutex::new(None);
+
+/// `None` schema is stored as the empty string so JSON keys are
+/// always concrete. This is the only callsite-facing translation.
+fn schema_key(schema: Option<&str>) -> String {
+    schema.unwrap_or("").to_string()
+}
+
+pub fn load(connection_id: Uuid, schema: Option<&str>, table: &str) -> FilterSet {
+    let mut guard = match CACHE.lock() {
+        Ok(g) => g,
+        Err(_) => return FilterSet::default(),
+    };
+    let map = guard.get_or_insert_with(load_from_disk);
+    map.get(&connection_id.to_string())
+        .and_then(|s| s.get(&schema_key(schema)))
+        .and_then(|t| t.get(table))
+        .cloned()
+        .unwrap_or_default()
+}
+
+pub fn save(connection_id: Uuid, schema: Option<&str>, table: &str, set: FilterSet) {
+    let mut guard = match CACHE.lock() {
+        Ok(g) => g,
+        Err(_) => return,
+    };
+    let map = guard.get_or_insert_with(load_from_disk);
+    if set.is_empty() {
+        // Empty FilterSet → remove the entry (and any now-empty
+        // ancestor maps) so the file shrinks back to its previous
+        // shape. Without this, "clear filter" would leave a
+        // `{"rules":[]}` blob behind on disk, which loads as an
+        // empty FilterSet anyway but bloats the file over time.
+        if let Some(schemas) = map.get_mut(&connection_id.to_string()) {
+            if let Some(tables) = schemas.get_mut(&schema_key(schema)) {
+                tables.remove(table);
+                if tables.is_empty() {
+                    schemas.remove(&schema_key(schema));
+                }
+            }
+            if schemas.is_empty() {
+                map.remove(&connection_id.to_string());
+            }
+        }
+    } else {
+        map.entry(connection_id.to_string())
+            .or_default()
+            .entry(schema_key(schema))
+            .or_default()
+            .insert(table.to_string(), set);
+    }
+    let snapshot = map.clone();
+    drop(guard);
+    relm4::spawn(async move {
+        if let Some(path) = xdg_config_path(FILE)
+            && let Err(e) = atomic_write_json(&path, &snapshot)
+        {
+            tracing::warn!(error = %e, "filter_settings: persist failed");
+        }
+    });
+}
+
+#[allow(dead_code)]
+pub fn clear(connection_id: Uuid, schema: Option<&str>, table: &str) {
+    // Public API even if no current callsite uses it directly —
+    // the dialog's "Clear all" button routes through `save` with an
+    // empty FilterSet which is the same code path. Kept exposed for
+    // a future "remove this filter" action elsewhere.
+    save(connection_id, schema, table, FilterSet::default());
+}
+
+fn load_from_disk() -> Connections {
+    let Some(path) = xdg_config_path(FILE) else {
+        return HashMap::new();
+    };
+    let Ok(bytes) = std::fs::read(path) else {
+        return HashMap::new();
+    };
+    serde_json::from_slice(&bytes).unwrap_or_default()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tablepro_core::{Combinator, FilterOp, FilterRule, FilterValue};
+
+    fn sample_set() -> FilterSet {
+        FilterSet {
+            combinator: Combinator::Or,
+            rules: vec![FilterRule {
+                column: "name".into(),
+                op: FilterOp::Eq,
+                value: Some(FilterValue::Single("alice".into())),
+            }],
+        }
+    }
+
+    #[test]
+    fn empty_set_returns_default() {
+        // Without touching the file, the cache initialises lazily; an
+        // entry that doesn't exist returns the FilterSet default.
+        let id = Uuid::new_v4();
+        let loaded = load(id, Some("public"), "users");
+        assert!(loaded.is_empty());
+    }
+
+    #[test]
+    fn schema_none_distinct_from_some() {
+        // An empty-string schema slot lives next to a "public" slot;
+        // they don't collide. Verified by writing two sets with
+        // different schema keys to the cache directly.
+        let mut connections: Connections = HashMap::new();
+        let id = Uuid::new_v4();
+        connections
+            .entry(id.to_string())
+            .or_default()
+            .entry(String::new())
+            .or_default()
+            .insert("t".into(), sample_set());
+        connections
+            .entry(id.to_string())
+            .or_default()
+            .entry("public".into())
+            .or_default()
+            .insert("t".into(), FilterSet::default());
+        let none_set = connections.get(&id.to_string()).and_then(|s| s.get("")).unwrap();
+        let public_set = connections.get(&id.to_string()).and_then(|s| s.get("public")).unwrap();
+        assert!(!none_set.get("t").unwrap().is_empty());
+        assert!(public_set.get("t").unwrap().is_empty());
+    }
+
+    #[test]
+    fn round_trip_serialises_and_deserialises() {
+        // Ensure the on-disk representation round-trips a non-trivial
+        // FilterSet — schemes / Combinator default behaviour / nested
+        // FilterValue tagged-content serde.
+        let original = sample_set();
+        let json = serde_json::to_string(&original).unwrap();
+        let parsed: FilterSet = serde_json::from_str(&json).unwrap();
+        assert_eq!(original, parsed);
+    }
+}

linux/crates/app/src/services/mod.rs

@@ -4,6 +4,7 @@ pub mod config_io;
 pub mod connection_monitor;
 pub mod connection_service;
 pub mod database_service;
+pub mod filter_settings;
 pub mod preferences;
 pub mod single_instance;
 pub mod structure_tracker;

linux/crates/app/src/ui/app/browse.rs

@@ -23,9 +23,11 @@ impl App {
     }
 
     /// Fire the SELECT * query for a specific browse tab. Result goes to
-    /// the same tab via `AppMsg::RowsLoaded(tab_id, ...)`.
+    /// the same tab via `AppMsg::RowsLoaded(tab_id, ...)`. Composes the
+    /// SELECT from the tab's current sort + filter + pagination state.
+    /// Filter and sort are server-side; pagination uses LIMIT/OFFSET.
     pub(super) fn fetch_browse_page(&self, tab_id: Uuid, sender: ComponentSender<Self>) {
-        let (schema, table, offset, limit, sort, driver_id) = {
+        let (schema, table, offset, limit, sort, filter, columns, driver_id) = {
             let tabs = self.workspace_tabs.borrow();
             let Some(controller) = tabs.get(&tab_id).and_then(|t| t.browse_controller()) else {
                 return;
@@ -37,6 +39,8 @@ impl App {
                 model.current_offset(),
                 model.page_size(),
                 model.current_sort(),
+                model.current_filter().clone(),
+                model.columns().to_vec(),
                 model.driver_id().to_string(),
             )
         };
@@ -46,37 +50,56 @@ impl App {
             return;
         };
         let order_by = sort.and_then(|(idx, asc)| {
-            let tabs = self.workspace_tabs.borrow();
-            let cols = tabs
-                .get(&tab_id)
-                .and_then(|t| t.browse_controller())
-                .map(|c| c.model().columns().to_vec())
-                .unwrap_or_default();
-            cols.get(idx).map(|c| {
+            columns.get(idx).map(|c| {
                 let name = tablepro_core::sql_dialect::quote_ident(&driver_id, &c.name);
                 let dir = if asc { "ASC" } else { "DESC" };
                 format!("{name} {dir}")
             })
         });
+
+        // Build WHERE up front so a filter validation error short-
+        // circuits to a toast without spawning the async command.
+        // Build returns None when the filter is empty; that path
+        // matches today's no-filter behaviour exactly.
+        let where_result = tablepro_core::build_filter_where(&driver_id, &columns, &filter);
+        let (where_sql, params) = match where_result {
+            Ok(Some((sql, p))) => (Some(sql), p),
+            Ok(None) => (None, Vec::new()),
+            Err(e) => {
+                sender.input(AppMsg::ShowToast(format!("{e}")));
+                return;
+            }
+        };
+
         let sender_clone = sender.clone();
         sender.command(move |_, shutdown| {
             shutdown
                 .register(async move {
-                    let result = match &order_by {
-                        Some(order) => {
-                            let qualified = match &schema {
-                                Some(s) => format!(
-                                    "{}.{}",
-                                    tablepro_core::sql_dialect::quote_ident(&driver_id, s),
-                                    tablepro_core::sql_dialect::quote_ident(&driver_id, &table)
-                                ),
-                                None => tablepro_core::sql_dialect::quote_ident(&driver_id, &table),
-                            };
-                            let sql =
-                                format!("SELECT * FROM {qualified} ORDER BY {order} LIMIT {limit} OFFSET {offset}");
-                            conn.query(&sql).await
+                    // No WHERE + no ORDER BY: keep the existing
+                    // fetch_rows fast-path so unchanged callers don't
+                    // pay the parametric overhead.
+                    let result = if where_sql.is_none() && order_by.is_none() {
+                        conn.fetch_rows(schema.as_deref(), &table, offset, limit).await
+                    } else {
+                        let qualified = match &schema {
+                            Some(s) => format!(
+                                "{}.{}",
+                                tablepro_core::sql_dialect::quote_ident(&driver_id, s),
+                                tablepro_core::sql_dialect::quote_ident(&driver_id, &table)
+                            ),
+                            None => tablepro_core::sql_dialect::quote_ident(&driver_id, &table),
+                        };
+                        let mut sql = format!("SELECT * FROM {qualified}");
+                        if let Some(w) = &where_sql {
+                            sql.push_str(" WHERE ");
+                            sql.push_str(w);
                         }
-                        None => conn.fetch_rows(schema.as_deref(), &table, offset, limit).await,
+                        if let Some(order) = &order_by {
+                            sql.push_str(" ORDER BY ");
+                            sql.push_str(order);
+                        }
+                        sql.push_str(&format!(" LIMIT {limit} OFFSET {offset}"));
+                        conn.query_params(&sql, &params).await
                     };
                     match result {
                         Ok(query_result) => sender_clone.input(AppMsg::RowsLoaded(tab_id, offset, query_result)),
@@ -116,7 +139,7 @@ impl App {
     }
 
     pub(super) fn fetch_browse_row_count(&self, tab_id: Uuid, sender: ComponentSender<Self>) {
-        let (schema, table, driver_id) = {
+        let (schema, table, filter, columns, driver_id) = {
             let tabs = self.workspace_tabs.borrow();
             let Some(controller) = tabs.get(&tab_id).and_then(|t| t.browse_controller()) else {
                 return;
@@ -125,13 +148,25 @@ impl App {
             (
                 model.schema().map(str::to_owned),
                 model.table().to_string(),
+                model.current_filter().clone(),
+                model.columns().to_vec(),
                 model.driver_id().to_string(),
             )
         };
 
         let Some(conn) = database_service::instance().active() else {
             return;
         };
+
+        // Same WHERE the page fetch uses, so the "of N" total matches
+        // the filtered result set. Validation errors are silently
+        // suppressed here — fetch_browse_page surfaces the toast for
+        // the same filter on the same tick, no need to double-toast.
+        let (where_sql, params) = match tablepro_core::build_filter_where(&driver_id, &columns, &filter) {
+            Ok(Some((sql, p))) => (Some(sql), p),
+            _ => (None, Vec::new()),
+        };
+
         let sender_clone = sender.clone();
         sender.command(move |_, shutdown| {
             shutdown
@@ -144,8 +179,17 @@ impl App {
                         ),
                         None => tablepro_core::sql_dialect::quote_ident(&driver_id, &table),
                     };
-                    let sql = format!("SELECT COUNT(*) FROM {qualified}");
-                    if let Ok(qr) = conn.query(&sql).await
+                    let mut sql = format!("SELECT COUNT(*) FROM {qualified}");
+                    if let Some(w) = &where_sql {
+                        sql.push_str(" WHERE ");
+                        sql.push_str(w);
+                    }
+                    let qr_result = if where_sql.is_some() {
+                        conn.query_params(&sql, &params).await
+                    } else {
+                        conn.query(&sql).await
+                    };
+                    if let Ok(qr) = qr_result
                         && let Some(row) = qr.rows.first()
                         && let Some(value) = row.first()
                     {
@@ -283,6 +327,41 @@ impl App {
         }
     }
 
+    /// Ctrl+R / Filter button — present the filter editor for the
+    /// active Browse tab. Hands the dialog the tab's columns +
+    /// current FilterSet; the dialog calls back via `on_apply` with
+    /// either the new set or `FilterSet::default()` for "Clear all".
+    /// Either way it routes through `BrowseTabInput::FilterApplied`
+    /// so persistence + chrome update + refetch all run in one place.
+    pub(super) fn on_show_filter_dialog(&self) {
+        let Some(id) = self.selected_browse_tab_id() else {
+            self.show_toast(&crate::tr!("Open a table to filter rows."));
+            return;
+        };
+        let (columns, current_filter) = {
+            let tabs = self.workspace_tabs.borrow();
+            let Some(controller) = tabs.get(&id).and_then(|t| t.browse_controller()) else {
+                return;
+            };
+            let model = controller.model();
+            (model.columns().to_vec(), model.current_filter().clone())
+        };
+        if columns.is_empty() {
+            // ColumnsLoaded hasn't fired yet — opening the dialog with
+            // no columns would just show an empty Add Rule list.
+            self.show_toast(&crate::tr!("Loading columns… try again in a moment."));
+            return;
+        }
+        let tab_id = id;
+        let workspace_tabs = self.workspace_tabs.clone();
+        let on_apply: std::rc::Rc<dyn Fn(tablepro_core::FilterSet)> = std::rc::Rc::new(move |set| {
+            if let Some(controller) = workspace_tabs.borrow().get(&tab_id).and_then(|t| t.browse_controller()) {
+                let _ = controller.sender().send(BrowseTabInput::FilterApplied(set));
+            }
+        });
+        crate::ui::filter_dialog::present(&self.window, columns, current_filter, on_apply);
+    }
+
     pub(super) fn on_refresh_active_tab(&self) {
         let Some(id) = self.selected_browse_tab_id() else {
             return;