feat: pre-fill row cells with filter data before row creation (#4854)

* feat: fill cells according to active filters

* chore: short circuit filter_row function

* chore: delete corresponding filters when filtered filter is deleted

* chore: validate filters when loading

* chore: remove unnecessary tuple in return

* chore: use trait

* chore: add tests

frontend/rust-lib/flowy-database2/src/entities/filter_entities/checklist_filter.rs

@@ -9,12 +9,11 @@ pub struct ChecklistFilterPB {
   pub condition: ChecklistFilterConditionPB,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, ProtoBuf_Enum)]
+#[derive(Debug, Clone, Default, PartialEq, Eq, ProtoBuf_Enum)]
 #[repr(u8)]
-#[derive(Default)]
 pub enum ChecklistFilterConditionPB {
-  IsComplete = 0,
   #[default]
+  IsComplete = 0,
   IsIncomplete = 1,
 }
 

frontend/rust-lib/flowy-database2/src/entities/filter_entities/relation_filter.rs

@@ -1,6 +1,7 @@
+use collab_database::{fields::Field, rows::Cell};
 use flowy_derive::ProtoBuf;
 
-use crate::services::filter::ParseFilterData;
+use crate::services::filter::{ParseFilterData, PreFillCellsWithFilter};
 
 #[derive(Eq, PartialEq, ProtoBuf, Debug, Default, Clone)]
 pub struct RelationFilterPB {
@@ -13,3 +14,9 @@ impl ParseFilterData for RelationFilterPB {
     RelationFilterPB { condition: 0 }
   }
 }
+
+impl PreFillCellsWithFilter for RelationFilterPB {
+  fn get_compliant_cell(&self, _field: &Field) -> (Option<Cell>, bool) {
+    (None, false)
+  }
+}

frontend/rust-lib/flowy-database2/src/services/database_view/view_editor.rs

@@ -148,7 +148,8 @@ impl DatabaseViewEditor {
     }
 
     // fill in cells according to active filters
-    // TODO(RS)
+    let filter_controller = self.filter_controller.clone();
+    filter_controller.fill_cells(&mut cells).await;
 
     result.collab_params.cells = cells;
 
@@ -739,6 +740,12 @@ impl DatabaseViewEditor {
   }
 
   pub async fn v_did_delete_field(&self, deleted_field_id: &str) {
+    let changeset = FilterChangeset::DeleteAllWithFieldId {
+      field_id: deleted_field_id.to_string(),
+    };
+    let notification = self.filter_controller.apply_changeset(changeset).await;
+    notify_did_update_filter(notification).await;
+
     let sorts = self.delegate.get_all_sorts(&self.view_id);
 
     if let Some(sort) = sorts.iter().find(|sort| sort.field_id == deleted_field_id) {
@@ -801,11 +808,10 @@ impl DatabaseViewEditor {
         .await;
 
       if old_field.field_type != field.field_type {
-        let filter_controller = self.filter_controller.clone();
         let changeset = FilterChangeset::DeleteAllWithFieldId {
           field_id: field.id.clone(),
         };
-        let notification = filter_controller.apply_changeset(changeset).await;
+        let notification = self.filter_controller.apply_changeset(changeset).await;
         notify_did_update_filter(notification).await;
       }
     }

frontend/rust-lib/flowy-database2/src/services/database_view/view_filter.rs

@@ -17,7 +17,6 @@ pub async fn make_filter_controller(
   notifier: DatabaseViewChangedNotifier,
   cell_cache: CellCache,
 ) -> Arc<FilterController> {
-  let filters = delegate.get_all_filters(view_id);
   let task_scheduler = delegate.get_task_scheduler();
   let filter_delegate = DatabaseViewFilterDelegateImpl(delegate.clone());
 
@@ -27,7 +26,6 @@ pub async fn make_filter_controller(
     &handler_id,
     filter_delegate,
     task_scheduler.clone(),
-    filters,
     cell_cache,
     notifier,
   )
@@ -62,6 +60,10 @@ impl FilterDelegate for DatabaseViewFilterDelegateImpl {
     self.0.get_row(view_id, rows_id)
   }
 
+  fn get_all_filters(&self, view_id: &str) -> Vec<Filter> {
+    self.0.get_all_filters(view_id)
+  }
+
   fn save_filters(&self, view_id: &str, filters: &[Filter]) {
     self.0.save_filters(view_id, filters)
   }

frontend/rust-lib/flowy-database2/src/services/field/type_options/checkbox_type_option/checkbox_filter.rs

@@ -1,4 +1,8 @@
+use collab_database::{fields::Field, rows::Cell};
+
 use crate::entities::{CheckboxCellDataPB, CheckboxFilterConditionPB, CheckboxFilterPB};
+use crate::services::cell::insert_checkbox_cell;
+use crate::services::filter::PreFillCellsWithFilter;
 
 impl CheckboxFilterPB {
   pub fn is_visible(&self, cell_data: &CheckboxCellDataPB) -> bool {
@@ -9,6 +13,20 @@ impl CheckboxFilterPB {
   }
 }
 
+impl PreFillCellsWithFilter for CheckboxFilterPB {
+  fn get_compliant_cell(&self, field: &Field) -> (Option<Cell>, bool) {
+    let is_checked = match self.condition {
+      CheckboxFilterConditionPB::IsChecked => Some(true),
+      CheckboxFilterConditionPB::IsUnChecked => None,
+    };
+
+    (
+      is_checked.map(|is_checked| insert_checkbox_cell(is_checked, field)),
+      false,
+    )
+  }
+}
+
 #[cfg(test)]
 mod tests {
   use crate::entities::{CheckboxCellDataPB, CheckboxFilterConditionPB, CheckboxFilterPB};

frontend/rust-lib/flowy-database2/src/services/field/type_options/checklist_type_option/checklist_filter.rs

@@ -1,5 +1,9 @@
+use collab_database::fields::Field;
+use collab_database::rows::Cell;
+
 use crate::entities::{ChecklistFilterConditionPB, ChecklistFilterPB};
 use crate::services::field::SelectOption;
+use crate::services::filter::PreFillCellsWithFilter;
 
 impl ChecklistFilterPB {
   pub fn is_visible(
@@ -37,3 +41,9 @@ impl ChecklistFilterPB {
     }
   }
 }
+
+impl PreFillCellsWithFilter for ChecklistFilterPB {
+  fn get_compliant_cell(&self, _field: &Field) -> (Option<Cell>, bool) {
+    (None, true)
+  }
+}

frontend/rust-lib/flowy-database2/src/services/field/type_options/date_type_option/date_filter.rs

@@ -1,8 +1,11 @@
 use crate::entities::{DateFilterConditionPB, DateFilterPB};
+use crate::services::cell::insert_date_cell;
+use crate::services::field::DateCellData;
+use crate::services::filter::PreFillCellsWithFilter;
 
-use chrono::{NaiveDate, NaiveDateTime};
-
-use super::DateCellData;
+use chrono::{Duration, NaiveDate, NaiveDateTime};
+use collab_database::fields::Field;
+use collab_database::rows::Cell;
 
 impl DateFilterPB {
   /// Returns `None` if the DateFilterPB doesn't have the necessary data for
@@ -95,6 +98,39 @@ impl DateFilterStrategy {
   }
 }
 
+impl PreFillCellsWithFilter for DateFilterPB {
+  fn get_compliant_cell(&self, field: &Field) -> (Option<Cell>, bool) {
+    let timestamp = match self.condition {
+      DateFilterConditionPB::DateIs
+      | DateFilterConditionPB::DateOnOrBefore
+      | DateFilterConditionPB::DateOnOrAfter => self.timestamp,
+      DateFilterConditionPB::DateBefore => self
+        .timestamp
+        .and_then(|timestamp| NaiveDateTime::from_timestamp_opt(timestamp, 0))
+        .map(|date_time| {
+          let answer = date_time - Duration::days(1);
+          answer.timestamp()
+        }),
+      DateFilterConditionPB::DateAfter => self
+        .timestamp
+        .and_then(|timestamp| NaiveDateTime::from_timestamp_opt(timestamp, 0))
+        .map(|date_time| {
+          let answer = date_time + Duration::days(1);
+          answer.timestamp()
+        }),
+      DateFilterConditionPB::DateWithIn => self.start,
+      _ => None,
+    };
+
+    let open_after_create = matches!(self.condition, DateFilterConditionPB::DateIsNotEmpty);
+
+    (
+      timestamp.map(|timestamp| insert_date_cell(timestamp, None, None, field)),
+      open_after_create,
+    )
+  }
+}
+
 #[cfg(test)]
 mod tests {
   use crate::entities::{DateFilterConditionPB, DateFilterPB};

frontend/rust-lib/flowy-database2/src/services/field/type_options/number_type_option/number_filter.rs

@@ -1,9 +1,13 @@
 use std::str::FromStr;
 
+use collab_database::fields::Field;
+use collab_database::rows::Cell;
 use rust_decimal::Decimal;
 
 use crate::entities::{NumberFilterConditionPB, NumberFilterPB};
+use crate::services::cell::insert_text_cell;
 use crate::services::field::NumberCellFormat;
+use crate::services::filter::PreFillCellsWithFilter;
 
 impl NumberFilterPB {
   pub fn is_visible(&self, cell_data: &NumberCellFormat) -> Option<bool> {
@@ -30,6 +34,39 @@ impl NumberFilterPB {
   }
 }
 
+impl PreFillCellsWithFilter for NumberFilterPB {
+  fn get_compliant_cell(&self, field: &Field) -> (Option<Cell>, bool) {
+    let expected_decimal = || Decimal::from_str(&self.content).ok();
+
+    let text = match self.condition {
+      NumberFilterConditionPB::Equal
+      | NumberFilterConditionPB::GreaterThanOrEqualTo
+      | NumberFilterConditionPB::LessThanOrEqualTo
+        if !self.content.is_empty() =>
+      {
+        Some(self.content.clone())
+      },
+      NumberFilterConditionPB::GreaterThan if !self.content.is_empty() => {
+        expected_decimal().map(|value| {
+          let answer = value + Decimal::from_f32_retain(1.0).unwrap();
+          answer.to_string()
+        })
+      },
+      NumberFilterConditionPB::LessThan if !self.content.is_empty() => {
+        expected_decimal().map(|value| {
+          let answer = value - Decimal::from_f32_retain(1.0).unwrap();
+          answer.to_string()
+        })
+      },
+      _ => None,
+    };
+
+    let open_after_create = matches!(self.condition, NumberFilterConditionPB::NumberIsNotEmpty);
+
+    // use `insert_text_cell` because self.content might not be a parsable i64.
+    (text.map(|s| insert_text_cell(s, field)), open_after_create)
+  }
+}
 enum NumberFilterStrategy {
   Equal(Decimal),
   NotEqual(Decimal),

frontend/rust-lib/flowy-database2/src/services/field/type_options/selection_type_option/select_filter.rs

@@ -1,5 +1,10 @@
+use collab_database::fields::Field;
+use collab_database::rows::Cell;
+
 use crate::entities::{SelectOptionFilterConditionPB, SelectOptionFilterPB};
-use crate::services::field::SelectOption;
+use crate::services::cell::insert_select_option_cell;
+use crate::services::field::{select_type_option_from_field, SelectOption};
+use crate::services::filter::PreFillCellsWithFilter;
 
 impl SelectOptionFilterPB {
   pub fn is_visible(&self, selected_options: &[SelectOption]) -> Option<bool> {
@@ -90,6 +95,40 @@ impl SelectOptionFilterStrategy {
   }
 }
 
+impl PreFillCellsWithFilter for SelectOptionFilterPB {
+  fn get_compliant_cell(&self, field: &Field) -> (Option<Cell>, bool) {
+    let get_non_empty_expected_options = || {
+      if !self.option_ids.is_empty() {
+        Some(self.option_ids.clone())
+      } else {
+        None
+      }
+    };
+
+    let option_ids = match self.condition {
+      SelectOptionFilterConditionPB::OptionIs => get_non_empty_expected_options(),
+      SelectOptionFilterConditionPB::OptionContains => {
+        get_non_empty_expected_options().map(|mut options| vec![options.swap_remove(0)])
+      },
+      SelectOptionFilterConditionPB::OptionIsNotEmpty => select_type_option_from_field(field)
+        .ok()
+        .map(|mut type_option| {
+          let options = type_option.mut_options();
+          if options.is_empty() {
+            vec![]
+          } else {
+            vec![options.swap_remove(0).id]
+          }
+        }),
+      _ => None,
+    };
+
+    (
+      option_ids.map(|ids| insert_select_option_cell(ids, field)),
+      false,
+    )
+  }
+}
 #[cfg(test)]
 mod tests {
   use crate::entities::{SelectOptionFilterConditionPB, SelectOptionFilterPB};

frontend/rust-lib/flowy-database2/src/services/field/type_options/text_type_option/text_filter.rs

@@ -1,4 +1,8 @@
+use collab_database::{fields::Field, rows::Cell};
+
 use crate::entities::{TextFilterConditionPB, TextFilterPB};
+use crate::services::cell::insert_text_cell;
+use crate::services::filter::PreFillCellsWithFilter;
 
 impl TextFilterPB {
   pub fn is_visible<T: AsRef<str>>(&self, cell_data: T) -> bool {
@@ -17,6 +21,26 @@ impl TextFilterPB {
   }
 }
 
+impl PreFillCellsWithFilter for TextFilterPB {
+  fn get_compliant_cell(&self, field: &Field) -> (Option<Cell>, bool) {
+    let text = match self.condition {
+      TextFilterConditionPB::TextIs
+      | TextFilterConditionPB::TextContains
+      | TextFilterConditionPB::TextStartsWith
+      | TextFilterConditionPB::TextEndsWith
+        if !self.content.is_empty() =>
+      {
+        Some(self.content.clone())
+      },
+      _ => None,
+    };
+
+    let open_after_create = matches!(self.condition, TextFilterConditionPB::TextIsNotEmpty);
+
+    (text.map(|s| insert_text_cell(s, field)), open_after_create)
+  }
+}
+
 #[cfg(test)]
 mod tests {
   #![allow(clippy::all)]

frontend/rust-lib/flowy-database2/src/services/field/type_options/type_option.rs

@@ -19,7 +19,7 @@ use crate::services::field::{
   CheckboxTypeOption, DateTypeOption, MultiSelectTypeOption, NumberTypeOption, RelationTypeOption,
   RichTextTypeOption, SingleSelectTypeOption, TimestampTypeOption, URLTypeOption,
 };
-use crate::services::filter::ParseFilterData;
+use crate::services::filter::{ParseFilterData, PreFillCellsWithFilter};
 use crate::services::sort::SortCondition;
 
 pub trait TypeOption {
@@ -58,7 +58,7 @@ pub trait TypeOption {
   type CellProtobufType: TryInto<Bytes, Error = ProtobufError> + Debug;
 
   /// Represents the filter configuration for this type option.
-  type CellFilter: ParseFilterData + Clone + Send + Sync + 'static;
+  type CellFilter: ParseFilterData + PreFillCellsWithFilter + Clone + Send + Sync + 'static;
 }
 /// This trait providing serialization and deserialization methods for cell data.
 ///

frontend/rust-lib/flowy-database2/src/services/filter/controller.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 
 use collab_database::database::gen_database_filter_id;
 use collab_database::fields::Field;
-use collab_database::rows::{Row, RowDetail, RowId};
+use collab_database::rows::{Cell, Cells, Row, RowDetail, RowId};
 use dashmap::DashMap;
 use serde::{Deserialize, Serialize};
 use tokio::sync::RwLock;
@@ -25,9 +25,14 @@ pub trait FilterDelegate: Send + Sync + 'static {
   fn get_fields(&self, view_id: &str, field_ids: Option<Vec<String>>) -> Fut<Vec<Field>>;
   fn get_rows(&self, view_id: &str) -> Fut<Vec<Arc<RowDetail>>>;
   fn get_row(&self, view_id: &str, rows_id: &RowId) -> Fut<Option<(usize, Arc<RowDetail>)>>;
+  fn get_all_filters(&self, view_id: &str) -> Vec<Filter>;
   fn save_filters(&self, view_id: &str, filters: &[Filter]);
 }
 
+pub trait PreFillCellsWithFilter {
+  fn get_compliant_cell(&self, field: &Field) -> (Option<Cell>, bool);
+}
+
 pub struct FilterController {
   view_id: String,
   handler_id: String,
@@ -51,13 +56,46 @@ impl FilterController {
     handler_id: &str,
     delegate: T,
     task_scheduler: Arc<RwLock<TaskDispatcher>>,
-    filters: Vec<Filter>,
     cell_cache: CellCache,
     notifier: DatabaseViewChangedNotifier,
   ) -> Self
   where
     T: FilterDelegate + 'static,
   {
+    // ensure every filter is valid
+    let field_ids = delegate
+      .get_fields(view_id, None)
+      .await
+      .into_iter()
+      .map(|field| field.id)
+      .collect::<Vec<_>>();
+
+    let mut need_save = false;
+
+    let mut filters = delegate.get_all_filters(view_id);
+    let mut filtering_field_ids: HashMap<String, Vec<String>> = HashMap::new();
+
+    for filter in filters.iter() {
+      filter.get_all_filtering_field_ids(&mut filtering_field_ids);
+    }
+
+    let mut delete_filter_ids = vec![];
+
+    for (field_id, filter_ids) in &filtering_field_ids {
+      if !field_ids.contains(field_id) {
+        need_save = true;
+        delete_filter_ids.extend(filter_ids);
+      }
+    }
+
+    for filter_id in delete_filter_ids {
+      Self::delete_filter(&mut filters, filter_id);
+    }
+
+    if need_save {
+      delegate.save_filters(view_id, &filters);
+    }
+
     Self {
       view_id: view_id.to_string(),
       handler_id: handler_id.to_string(),
@@ -116,106 +154,6 @@ impl FilterController {
     });
   }
 
-  async fn get_field_map(&self) -> HashMap<String, Field> {
-    self
-      .delegate
-      .get_fields(&self.view_id, None)
-      .await
-      .into_iter()
-      .map(|field| (field.id.clone(), field))
-      .collect::<HashMap<String, Field>>()
-  }
-
-  #[tracing::instrument(
-    name = "process_filter_task",
-    level = "trace",
-    skip_all,
-    fields(filter_result),
-    err
-  )]
-  pub async fn process(&self, predicate: &str) -> FlowyResult<()> {
-    let event_type = FilterEvent::from_str(predicate).unwrap();
-    match event_type {
-      FilterEvent::FilterDidChanged => self.filter_all_rows().await?,
-      FilterEvent::RowDidChanged(row_id) => self.filter_single_row(row_id).await?,
-    }
-    Ok(())
-  }
-
-  async fn filter_single_row(&self, row_id: RowId) -> FlowyResult<()> {
-    let filters = self.filters.read().await;
-
-    if let Some((_, row_detail)) = self.delegate.get_row(&self.view_id, &row_id).await {
-      let field_by_field_id = self.get_field_map().await;
-      let mut notification = FilterResultNotification::new(self.view_id.clone());
-      if let Some((row_id, is_visible)) = filter_row(
-        &row_detail.row,
-        &self.result_by_row_id,
-        &field_by_field_id,
-        &self.cell_cache,
-        &filters,
-      ) {
-        if is_visible {
-          if let Some((index, _row)) = self.delegate.get_row(&self.view_id, &row_id).await {
-            notification.visible_rows.push(
-              InsertedRowPB::new(RowMetaPB::from(row_detail.as_ref())).with_index(index as i32),
-            )
-          }
-        } else {
-          notification.invisible_rows.push(row_id);
-        }
-      }
-
-      let _ = self
-        .notifier
-        .send(DatabaseViewChanged::FilterNotification(notification));
-    }
-    Ok(())
-  }
-
-  async fn filter_all_rows(&self) -> FlowyResult<()> {
-    let filters = self.filters.read().await;
-
-    let field_by_field_id = self.get_field_map().await;
-    let mut visible_rows = vec![];
-    let mut invisible_rows = vec![];
-
-    for (index, row_detail) in self
-      .delegate
-      .get_rows(&self.view_id)
-      .await
-      .into_iter()
-      .enumerate()
-    {
-      if let Some((row_id, is_visible)) = filter_row(
-        &row_detail.row,
-        &self.result_by_row_id,
-        &field_by_field_id,
-        &self.cell_cache,
-        &filters,
-      ) {
-        if is_visible {
-          let row_meta = RowMetaPB::from(row_detail.as_ref());
-          visible_rows.push(InsertedRowPB::new(row_meta).with_index(index as i32))
-        } else {
-          invisible_rows.push(row_id);
-        }
-      }
-    }
-
-    let notification = FilterResultNotification {
-      view_id: self.view_id.clone(),
-      invisible_rows,
-      visible_rows,
-    };
-    tracing::trace!("filter result {:?}", filters);
-    let _ = self
-      .notifier
-      .send(DatabaseViewChanged::FilterNotification(notification));
-
-    Ok(())
-  }
-
   pub async fn did_receive_row_changed(&self, row_id: RowId) {
     if !self.filters.read().await.is_empty() {
       self
@@ -281,38 +219,14 @@ impl FilterController {
       FilterChangeset::Delete {
         filter_id,
         field_id: _,
-      } => {
-        for (position, filter) in filters.iter_mut().enumerate() {
-          if filter.id == filter_id {
-            filters.remove(position);
-            break;
-          }
-          let parent_filter = filter.find_parent_of_filter(&filter_id);
-          if let Some(filter) = parent_filter {
-            let result = filter.delete_filter(&filter_id);
-            if result.is_ok() {
-              break;
-            }
-          }
-        }
-      },
+      } => Self::delete_filter(&mut filters, &filter_id),
       FilterChangeset::DeleteAllWithFieldId { field_id } => {
-        let mut filter_ids: Vec<String> = vec![];
-        for filter in filters.iter_mut() {
+        let mut filter_ids = vec![];
+        for filter in filters.iter() {
           filter.find_all_filters_with_field_id(&field_id, &mut filter_ids);
         }
-
         for filter_id in filter_ids {
-          for (position, filter) in filters.iter_mut().enumerate() {
-            if filter.id == filter_id {
-              filters.remove(position);
-              break;
-            }
-            let parent_filter = filter.find_parent_of_filter(&filter_id);
-            if let Some(filter) = parent_filter {
-              let _ = filter.delete_filter(&filter_id);
-            }
-          }
+          Self::delete_filter(&mut filters, &filter_id)
         }
       },
     }
@@ -325,6 +239,210 @@ impl FilterController {
 
     FilterChangesetNotificationPB::from_filters(&self.view_id, &filters)
   }
+
+  pub async fn fill_cells(&self, cells: &mut Cells) -> bool {
+    let filters = self.filters.read().await;
+
+    let mut open_after_create = false;
+
+    let mut min_required_filters: Vec<&FilterInner> = vec![];
+    for filter in filters.iter() {
+      filter.get_min_effective_filters(&mut min_required_filters);
+    }
+
+    let field_map = self.get_field_map().await;
+
+    while let Some(current_inner) = min_required_filters.pop() {
+      if let FilterInner::Data {
+        field_id,
+        field_type,
+        condition_and_content,
+      } = &current_inner
+      {
+        if min_required_filters.iter().any(
+          |inner| matches!(inner, FilterInner::Data { field_id: other_id, .. } if other_id == field_id),
+        ) {
+          min_required_filters.retain(
+            |inner| matches!(inner, FilterInner::Data { field_id: other_id, .. } if other_id != field_id),
+          );
+          open_after_create = true;
+          continue;
+        }
+
+        if let Some(field) = field_map.get(field_id) {
+          let (cell, flag) = match field_type {
+            FieldType::RichText | FieldType::URL => {
+              let filter = condition_and_content.cloned::<TextFilterPB>().unwrap();
+              filter.get_compliant_cell(field)
+            },
+            FieldType::Number => {
+              let filter = condition_and_content.cloned::<NumberFilterPB>().unwrap();
+              filter.get_compliant_cell(field)
+            },
+            FieldType::DateTime => {
+              let filter = condition_and_content.cloned::<DateFilterPB>().unwrap();
+              filter.get_compliant_cell(field)
+            },
+            FieldType::SingleSelect => {
+              let filter = condition_and_content
+                .cloned::<SelectOptionFilterPB>()
+                .unwrap();
+              filter.get_compliant_cell(field)
+            },
+            FieldType::MultiSelect => {
+              let filter = condition_and_content
+                .cloned::<SelectOptionFilterPB>()
+                .unwrap();
+              filter.get_compliant_cell(field)
+            },
+            FieldType::Checkbox => {
+              let filter = condition_and_content.cloned::<CheckboxFilterPB>().unwrap();
+              filter.get_compliant_cell(field)
+            },
+            FieldType::Checklist => {
+              let filter = condition_and_content.cloned::<ChecklistFilterPB>().unwrap();
+              filter.get_compliant_cell(field)
+            },
+            _ => (None, false),
+          };
+
+          if let Some(cell) = cell {
+            cells.insert(field_id.clone(), cell);
+          }
+
+          if flag {
+            open_after_create = flag;
+          }
+        }
+      }
+    }
+
+    open_after_create
+  }
+
+  #[tracing::instrument(
+    name = "process_filter_task",
+    level = "trace",
+    skip_all,
+    fields(filter_result),
+    err
+  )]
+  pub async fn process(&self, predicate: &str) -> FlowyResult<()> {
+    let event_type = FilterEvent::from_str(predicate).unwrap();
+    match event_type {
+      FilterEvent::FilterDidChanged => self.filter_all_rows_handler().await?,
+      FilterEvent::RowDidChanged(row_id) => self.filter_single_row_handler(row_id).await?,
+    }
+    Ok(())
+  }
+
+  async fn filter_single_row_handler(&self, row_id: RowId) -> FlowyResult<()> {
+    let filters = self.filters.read().await;
+
+    if let Some((_, row_detail)) = self.delegate.get_row(&self.view_id, &row_id).await {
+      let field_by_field_id = self.get_field_map().await;
+      let mut notification = FilterResultNotification::new(self.view_id.clone());
+      if let Some(is_visible) = filter_row(
+        &row_detail.row,
+        &self.result_by_row_id,
+        &field_by_field_id,
+        &self.cell_cache,
+        &filters,
+      ) {
+        if is_visible {
+          if let Some((index, _row)) = self.delegate.get_row(&self.view_id, &row_id).await {
+            notification.visible_rows.push(
+              InsertedRowPB::new(RowMetaPB::from(row_detail.as_ref())).with_index(index as i32),
+            )
+          }
+        } else {
+          notification.invisible_rows.push(row_id);
+        }
+      }
+
+      let _ = self
+        .notifier
+        .send(DatabaseViewChanged::FilterNotification(notification));
+    }
+    Ok(())
+  }
+
+  async fn filter_all_rows_handler(&self) -> FlowyResult<()> {
+    let filters = self.filters.read().await;
+
+    let field_by_field_id = self.get_field_map().await;
+    let mut visible_rows = vec![];
+    let mut invisible_rows = vec![];
+
+    for (index, row_detail) in self
+      .delegate
+      .get_rows(&self.view_id)
+      .await
+      .into_iter()
+      .enumerate()
+    {
+      if let Some(is_visible) = filter_row(
+        &row_detail.row,
+        &self.result_by_row_id,
+        &field_by_field_id,
+        &self.cell_cache,
+        &filters,
+      ) {
+        if is_visible {
+          let row_meta = RowMetaPB::from(row_detail.as_ref());
+          visible_rows.push(InsertedRowPB::new(row_meta).with_index(index as i32))
+        } else {
+          invisible_rows.push(row_detail.row.id.clone());
+        }
+      }
+    }
+
+    let notification = FilterResultNotification {
+      view_id: self.view_id.clone(),
+      invisible_rows,
+      visible_rows,
+    };
+    tracing::trace!("filter result {:?}", filters);
+    let _ = self
+      .notifier
+      .send(DatabaseViewChanged::FilterNotification(notification));
+
+    Ok(())
+  }
+
+  async fn get_field_map(&self) -> HashMap<String, Field> {
+    self
+      .delegate
+      .get_fields(&self.view_id, None)
+      .await
+      .into_iter()
+      .map(|field| (field.id.clone(), field))
+      .collect::<HashMap<String, Field>>()
+  }
+
+  fn delete_filter(filters: &mut Vec<Filter>, filter_id: &str) {
+    let mut find_root_filter: Option<usize> = None;
+    let mut find_parent_of_non_root_filter: Option<&mut Filter> = None;
+
+    for (position, filter) in filters.iter_mut().enumerate() {
+      if filter.id == filter_id {
+        find_root_filter = Some(position);
+        break;
+      }
+      if let Some(filter) = filter.find_parent_of_filter(filter_id) {
+        find_parent_of_non_root_filter = Some(filter);
+        break;
+      }
+    }
+
+    if let Some(pos) = find_root_filter {
+      filters.remove(pos);
+    } else if let Some(filter) = find_parent_of_non_root_filter {
+      if let Err(err) = filter.delete_filter(filter_id) {
+        tracing::error!("error while deleting filter: {}", err);
+      }
+    }
+  }
 }
 
 /// Returns `Some` if the visibility of the row changed after applying the filter and `None`
@@ -336,22 +454,28 @@ fn filter_row(
   field_by_field_id: &HashMap<String, Field>,
   cell_data_cache: &CellCache,
   filters: &Vec<Filter>,
-) -> Option<(RowId, bool)> {
+) -> Option<bool> {
   // Create a filter result cache if it doesn't exist
   let mut filter_result = result_by_row_id.entry(row.id.clone()).or_insert(true);
   let old_is_visible = *filter_result;
 
   let mut new_is_visible = true;
+
   for filter in filters {
     if let Some(is_visible) = apply_filter(row, field_by_field_id, cell_data_cache, filter) {
       new_is_visible = new_is_visible && is_visible;
+
+      // short-circuit as soon as one filter tree returns false
+      if !new_is_visible {
+        break;
+      }
     }
   }
 
   *filter_result = new_is_visible;
 
   if old_is_visible != new_is_visible {
-    Some((row.id.clone(), new_is_visible))
+    Some(new_is_visible)
   } else {
     None
   }

frontend/rust-lib/flowy-database2/src/services/filter/entities.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::mem;
 
 use anyhow::bail;
@@ -25,7 +26,8 @@ pub struct Filter {
 }
 
 impl Filter {
-  /// Recursively determine whether there are any data filters in the filter tree.
+  /// Recursively determine whether there are any data filters in the filter tree. A tree that has
+  /// multiple AND/OR filters but no Data filters is considered "empty".
   pub fn is_empty(&self) -> bool {
     match &self.inner {
       FilterInner::And { children } | FilterInner::Or { children } => children
@@ -36,6 +38,7 @@ impl Filter {
     }
   }
 
+  /// Recursively find a filter based on `filter_id`. Returns `None` if the filter cannot be found.
   pub fn find_filter(&mut self, filter_id: &str) -> Option<&mut Self> {
     if self.id == filter_id {
       return Some(self);
@@ -54,6 +57,8 @@ impl Filter {
     }
   }
 
+  /// Recursively find the parent of a filter whose id is `filter_id`. Returns `None` if the filter
+  /// cannot be found.
   pub fn find_parent_of_filter(&mut self, filter_id: &str) -> Option<&mut Self> {
     if self.id == filter_id {
       return None;
@@ -75,7 +80,7 @@ impl Filter {
     }
   }
 
-  /// converts a filter from And/Or/Data to And/Or. If the current type of the filter is Data,
+  /// Converts a filter from And/Or/Data to And/Or. If the current type of the filter is Data,
   /// return the FilterInner after the conversion.
   pub fn convert_to_and_or_filter_type(
     &mut self,
@@ -118,6 +123,10 @@ impl Filter {
     }
   }
 
+  /// Insert a filter into the current filter in the filter tree. If the current filter
+  /// is an AND/OR filter, then the filter is appended to its children. Otherwise, the current
+  /// filter is converted to an AND filter, after which the current data filter and the new filter
+  /// are added to the AND filter's children.
   pub fn insert_filter(&mut self, filter: Filter) -> FlowyResult<()> {
     match &mut self.inner {
       FilterInner::And { children } | FilterInner::Or { children } => {
@@ -141,6 +150,8 @@ impl Filter {
     Ok(())
   }
 
+  /// Update the criteria of a data filter. Return an error if the current filter is an AND/OR
+  /// filter.
   pub fn update_filter_data(&mut self, filter_data: FilterInner) -> FlowyResult<()> {
     match &self.inner {
       FilterInner::And { .. } | FilterInner::Or { .. } => Err(FlowyError::internal().with_context(
@@ -153,6 +164,9 @@ impl Filter {
     }
   }
 
+  /// Delete a filter based on `filter_id`. The current filter must be the parent of the filter
+  /// whose id is `filter_id`. Returns an error if the current filter is a Data filter (which
+  /// cannot have children), or the filter to be deleted cannot be found.
   pub fn delete_filter(&mut self, filter_id: &str) -> FlowyResult<()> {
     match &mut self.inner {
       FilterInner::And { children } | FilterInner::Or { children } => children
@@ -171,24 +185,63 @@ impl Filter {
     }
   }
 
-  pub fn find_all_filters_with_field_id(&mut self, matching_field_id: &str, ids: &mut Vec<String>) {
-    match &mut self.inner {
+  /// Recursively finds any Data filter whose `field_id` is equal to `matching_field_id`. Any found
+  /// filters' id is appended to the `ids` vector.
+  pub fn find_all_filters_with_field_id(&self, matching_field_id: &str, ids: &mut Vec<String>) {
+    match &self.inner {
       FilterInner::And { children } | FilterInner::Or { children } => {
-        for child_filter in children.iter_mut() {
+        for child_filter in children.iter() {
           child_filter.find_all_filters_with_field_id(matching_field_id, ids);
         }
       },
-      FilterInner::Data {
-        field_id,
-        field_type: _,
-        condition_and_content: _,
-      } => {
+      FilterInner::Data { field_id, .. } => {
         if field_id == matching_field_id {
           ids.push(self.id.clone());
         }
       },
     }
   }
+
+  /// Recursively determine the smallest set of filters that loosely represents the filter tree. The
+  /// filters are appended to the `min_effective_filters` vector. The following rules are followed
+  /// when determining if a filter should get included. If the current filter is:
+  ///
+  /// 1. a Data filter, then it should be included.
+  /// 2. an AND filter, then all of its effective children should be
+  /// included.
+  /// 3. an OR filter, then only the first child should be included.
+  pub fn get_min_effective_filters<'a>(&'a self, min_effective_filters: &mut Vec<&'a FilterInner>) {
+    match &self.inner {
+      FilterInner::And { children } => {
+        for filter in children.iter() {
+          filter.get_min_effective_filters(min_effective_filters);
+        }
+      },
+      FilterInner::Or { children } => {
+        if let Some(filter) = children.first() {
+          filter.get_min_effective_filters(min_effective_filters);
+        }
+      },
+      FilterInner::Data { .. } => min_effective_filters.push(&self.inner),
+    }
+  }
+
+  /// Recursively get all of the filtering field ids and the associated filter_ids
+  pub fn get_all_filtering_field_ids(&self, field_ids: &mut HashMap<String, Vec<String>>) {
+    match &self.inner {
+      FilterInner::And { children } | FilterInner::Or { children } => {
+        for child in children.iter() {
+          child.get_all_filtering_field_ids(field_ids);
+        }
+      },
+      FilterInner::Data { field_id, .. } => {
+        field_ids
+          .entry(field_id.clone())
+          .and_modify(|filter_ids| filter_ids.push(self.id.clone()))
+          .or_insert_with(|| vec![self.id.clone()]);
+      },
+    }
+  }
 }
 
 #[derive(Debug)]

frontend/rust-lib/flowy-database2/src/services/sort/controller.rs

@@ -117,6 +117,7 @@ impl SortController {
   pub async fn process(&mut self, predicate: &str) -> FlowyResult<()> {
     let event_type = SortEvent::from_str(predicate).unwrap();
     let mut row_details = self.delegate.get_rows(&self.view_id).await;
+
     match event_type {
       SortEvent::SortDidChanged | SortEvent::DeleteAllSorts => {
         self.sort_rows(&mut row_details).await;