diff --git a/book/src/repo-list-for-filters.md b/book/src/repo-list-for-filters.md index a56ef7d..335a2a1 100644 --- a/book/src/repo-list-for-filters.md +++ b/book/src/repo-list-for-filters.md @@ -48,7 +48,23 @@ let filters = UserDocumentFilters { ### Per-Sort-Column Functions -For each `list_by` column, a `list_for_filters_by_{sort_col}` function is generated with SQL that uses nullable WHERE patterns: +For each `list_by` column, a `list_for_filters_by_{sort_col}` function is generated. Instead of one catch-all query, it contains one static SQL query per **filter combination × cursor state** and dispatches at runtime on which filters are `Some`. Present filters compile to plain, index-friendly (sargable) predicates; absent filters are omitted entirely: + +```sql +-- user_id = Some(..), status = None, first page +SELECT id FROM user_documents + WHERE user_id = $1 + ORDER BY id ASC LIMIT $2 + +-- both filters set, paginating from a cursor +SELECT id FROM user_documents + WHERE user_id = $1 AND status = $2 AND (id > $4) + ORDER BY id ASC LIMIT $3 +``` + +This matters for performance: the planner can turn `col = $k` into an index condition, which is impossible through a `COALESCE(col = $k, $k IS NULL)` catch-all (a single generic plan must serve both `NULL` and non-`NULL` parameters, so the predicate never becomes an index qual and every call full-scans the table). + +For entities with more than 4 `list_for` columns the combination matrix is capped: only the no-filter, single-filter, and all-filters combinations get specialized queries, and remaining combinations fall back to the catch-all COALESCE-based SQL (correct, just not sargable): ```sql SELECT id FROM user_documents @@ -58,15 +74,13 @@ SELECT id FROM user_documents ORDER BY id ASC LIMIT $3 ``` -When a parameter is `NULL` (i.e., `None`), the `COALESCE` evaluates to `true`, effectively skipping that filter. - ### A Dispatch Function The `list_for_filters` function matches on the sort column and intelligently delegates to the most efficient underlying function: - **No filters set** (`Filters::default()`): proxies to `list_by_{sort}` (simple query, full index usage) - **Exactly one filter set**: proxies to `list_for_{col}_by_{sort}` (single-column WHERE, full index usage) -- **Two or more filters set**: uses the per-sort COALESCE-based SQL (multi-column nullable WHERE) +- **Two or more filters set**: uses the per-sort specialized query matching the exact filter combination (sargable), falling back to the COALESCE-based SQL only for combinations beyond the specialization cap ## Important Notes diff --git a/es-entity-macros/src/repo/list_by_fn.rs b/es-entity-macros/src/repo/list_by_fn.rs index ae8b2b0..e1cbf5c 100644 --- a/es-entity-macros/src/repo/list_by_fn.rs +++ b/es-entity-macros/src/repo/list_by_fn.rs @@ -5,6 +5,59 @@ use quote::{TokenStreamExt, quote}; use super::options::*; +/// Cursor pagination states that each get their own SQL text, so that no +/// variant needs the non-sargable `COALESCE(..., $ IS NULL)` catch-all. +/// +/// The generated list fns dispatch on these states at runtime (on the +/// `Some`-ness of the destructured cursor values) and every emitted query is +/// a static `es_query!` literal — compile-time checked and index-friendly. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CursorState { + /// Page 1 (no cursor): no cursor predicate at all — the query rides the + /// index ordering with an early-exit `LIMIT`. + First, + /// Cursor present on a non-NULL sort value: bare `(col, id)` row + /// comparison — sargable against a composite index. + After, + /// Cursor present on a NULL sort value (only possible for `Option` + /// sort columns): explicit NULL-aware predicate. + AfterNull, + /// Cursor present, but whether the sort value encodes as SQL NULL is + /// undetectable from Rust (non-`Option` type annotated `nullable`, where + /// a custom `sqlx::Encode` may write NULL): the query must handle both + /// cases, so it keeps the all-cases COALESCE fallback predicate + /// (correct, not sargable). + AfterMaybeNull, +} + +/// Assemble a `SELECT ... [WHERE ...] ORDER BY ... LIMIT $n` query string +/// from individual predicates. +pub fn assemble_select( + select_columns: &str, + table_name: &str, + conditions: &[String], + order_by: &str, + limit_param_idx: u32, +) -> String { + let mut query = format!("SELECT {select_columns} FROM {table_name}"); + if !conditions.is_empty() { + query.push_str(" WHERE "); + query.push_str(&conditions.join(" AND ")); + } + query.push_str(&format!(" ORDER BY {order_by} LIMIT ${limit_param_idx}")); + query +} + +/// The `deleted = FALSE` predicate for repos with soft delete, on the +/// non-`include_deleted` fn variants. +pub fn not_deleted_predicate(delete: DeleteOption) -> Option { + if delete.is_soft() { + Some("deleted = FALSE".to_string()) + } else { + None + } +} + pub struct CursorStruct<'a> { pub id: &'a syn::Ident, pub entity: &'a syn::Ident, @@ -109,19 +162,148 @@ impl CursorStruct<'_> { } } - pub fn query_arg_tokens(&self) -> TokenStream { + /// The cursor states this sort column needs distinct SQL for. + pub fn cursor_states(&self) -> &'static [CursorState] { + if self.column.is_id() || !self.column.is_nullable_column() { + &[CursorState::First, CursorState::After] + } else if self.column.is_optional() { + &[ + CursorState::First, + CursorState::After, + CursorState::AfterNull, + ] + } else { + // `nullable`-annotated non-Option type: NULL-ness of the cursor + // value is invisible to Rust, so the cursor-present variant must + // keep the all-cases fallback predicate. + &[CursorState::First, CursorState::AfterMaybeNull] + } + } + + /// The cursor predicate for a specialized state, or `None` for + /// [`CursorState::First`] (page 1 needs no predicate). + /// + /// `offset` is the number of query parameters preceding the `LIMIT` + /// parameter (i.e. LIMIT lands on `$(offset + 1)`). + /// + /// The specialized forms are sargable: a bare `(col, id)` row comparison + /// is an index qual against a composite index, unlike the + /// `COALESCE((col, id) < ($c, $i), $i IS NULL)` catch-all which defeats + /// index extraction. The NULL-cursor forms replicate the exact edge + /// semantics documented on [`Self::condition`]: + /// + /// - ASC (NULLS FIRST), cursor on a NULL row: all non-NULL rows plus + /// NULL rows with a greater id come "after" → `col IS NOT NULL OR id > + /// $i`. + /// - DESC (NULLS LAST), cursor on a NULL row: only NULL rows with a + /// smaller id come after → `col IS NULL AND id < $i`. + /// - DESC (NULLS LAST), cursor on a non-NULL row: NULL rows sort last, + /// so they are all still "after" → `col IS NULL OR (col, id) < ($c, + /// $i)`. + pub fn condition_for_state( + &self, + state: CursorState, + offset: u32, + ascending: bool, + ) -> Option { + let comp = if ascending { ">" } else { "<" }; + let id_offset = offset + 2; + let column_offset = offset + 3; + + match state { + CursorState::First => None, + CursorState::AfterMaybeNull => Some(self.condition(offset, ascending)), + CursorState::After => { + if self.column.is_id() { + Some(format!("id {comp} ${id_offset}")) + } else if !self.column.is_nullable_column() { + Some(format!( + "({0}, id) {comp} (${column_offset}, ${id_offset})", + self.column.name() + )) + } else if ascending { + Some(format!( + "({0}, id) > (${column_offset}, ${id_offset})", + self.column.name() + )) + } else { + Some(format!( + "({0} IS NULL OR ({0}, id) < (${column_offset}, ${id_offset}))", + self.column.name() + )) + } + } + CursorState::AfterNull => { + if ascending { + Some(format!( + "({0} IS NOT NULL OR id > ${id_offset})", + self.column.name() + )) + } else { + Some(format!( + "({0} IS NULL AND id < ${id_offset})", + self.column.name() + )) + } + } + } + } + + /// Scrutinee elements (one or two bool expressions over the destructured + /// cursor locals) identifying the cursor state at runtime. + pub fn state_scrutinee_elems(&self) -> Vec { + if self.column.is_nullable_column() && self.column.is_optional() { + let column_name = self.column.name(); + vec![quote! { id.is_some() }, quote! { #column_name.is_some() }] + } else { + vec![quote! { id.is_none() }] + } + } + + /// Pattern elements matching [`Self::state_scrutinee_elems`] for one + /// state. + pub fn state_pattern_elems(&self, state: CursorState) -> Vec { + if self.column.is_nullable_column() && self.column.is_optional() { + match state { + CursorState::First => vec![quote! { false }, quote! { _ }], + CursorState::After => vec![quote! { true }, quote! { true }], + CursorState::AfterNull => vec![quote! { true }, quote! { false }], + CursorState::AfterMaybeNull => { + unreachable!("Option columns never use AfterMaybeNull") + } + } + } else { + match state { + CursorState::First => vec![quote! { true }], + _ => vec![quote! { false }], + } + } + } + + /// Cursor value bindings (without the `LIMIT` binding) for a state. + pub fn cursor_arg_tokens_for_state(&self, state: CursorState) -> TokenStream { + let id = self.id; + + match state { + CursorState::First => quote! {}, + CursorState::AfterNull => quote! { + id as Option<#id>, + }, + _ => self.cursor_arg_tokens(), + } + } + + fn cursor_arg_tokens(&self) -> TokenStream { let id = self.id; if self.column.is_id() { quote! { - (first + 1) as i64, id as Option<#id>, } } else if self.column.is_optional() { let column_name = self.column.name(); let column_type = self.column.ty(); quote! { - (first + 1) as i64, id as Option<#id>, #column_name as #column_type, } @@ -129,13 +311,20 @@ impl CursorStruct<'_> { let column_name = self.column.name(); let column_type = self.column.ty(); quote! { - (first + 1) as i64, id as Option<#id>, #column_name as Option<#column_type>, } } } + pub fn query_arg_tokens(&self) -> TokenStream { + let cursor_args = self.cursor_arg_tokens(); + quote! { + (first + 1) as i64, + #cursor_args + } + } + pub fn destructure_tokens(&self) -> TokenStream { let column_name = self.column.name(); @@ -302,7 +491,6 @@ impl ToTokens for ListByFn<'_> { let destructure_tokens = self.cursor().destructure_tokens(); let select_columns = cursor.select_columns(None); - let arg_tokens = cursor.query_arg_tokens(); for delete in [DeleteOption::No, DeleteOption::Soft] { let fn_name = syn::Ident::new( @@ -322,76 +510,73 @@ impl ToTokens for ListByFn<'_> { Span::call_site(), ); - let asc_query = format!( - r#"SELECT {} FROM {} WHERE ({}){} ORDER BY {} LIMIT $1"#, - select_columns, - self.table_name, - cursor.condition(0, true), - if delete == DeleteOption::No { - self.delete.not_deleted_condition() - } else { - "" - }, - cursor.order_by(true), - ); - let desc_query = format!( - r#"SELECT {} FROM {} WHERE ({}){} ORDER BY {} LIMIT $1"#, - select_columns, - self.table_name, - cursor.condition(0, false), - if delete == DeleteOption::No { - self.delete.not_deleted_condition() - } else { - "" - }, - cursor.order_by(false), - ); - let forgettable_tbl_arg = if let Some(tbl) = self.forgettable_table_name { quote! { forgettable_tbl = #tbl, } } else { quote! {} }; - let es_query_asc_call = if let Some(prefix) = self.ignore_prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #asc_query, - #arg_tokens - ) - } - } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #asc_query, - #arg_tokens - ) + let make_es_query = |query: &str, args: &TokenStream| -> TokenStream { + if let Some(prefix) = self.ignore_prefix { + quote! { + es_entity::es_query!( + tbl_prefix = #prefix, + #forgettable_tbl_arg + #query, + #args + ) + } + } else { + quote! { + es_entity::es_query!( + entity = #entity, + #forgettable_tbl_arg + #query, + #args + ) + } } }; - let es_query_desc_call = if let Some(prefix) = self.ignore_prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #desc_query, - #arg_tokens - ) - } - } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #desc_query, - #arg_tokens - ) + let mut query_arms = TokenStream::new(); + for state in cursor.cursor_states() { + for ascending in [true, false] { + let mut conditions: Vec = Vec::new(); + if let Some(condition) = cursor.condition_for_state(*state, 0, ascending) { + conditions.push(format!("({condition})")); + } + if delete == DeleteOption::No + && let Some(not_deleted) = not_deleted_predicate(self.delete) + { + conditions.push(not_deleted); + } + let query = assemble_select( + &select_columns, + self.table_name, + &conditions, + &cursor.order_by(ascending), + 1, + ); + let cursor_args = cursor.cursor_arg_tokens_for_state(*state); + let args = quote! { + (first + 1) as i64, + #cursor_args + }; + let es_query_call = make_es_query(&query, &args); + let direction_pattern = if ascending { + quote! { es_entity::ListDirection::Ascending } + } else { + quote! { es_entity::ListDirection::Descending } + }; + let state_pattern = cursor.state_pattern_elems(*state); + query_arms.append_all(quote! { + (#direction_pattern, #(#state_pattern),*) => { + #es_query_call.fetch_n(op, first).await? + }, + }); } - }; + } + let cursor_state_scrutinee = cursor.state_scrutinee_elems(); #[cfg(feature = "instrument")] let ( @@ -473,13 +658,8 @@ impl ToTokens for ListByFn<'_> { #destructure_tokens #record_fields - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { - #es_query_asc_call.fetch_n(op, first).await? - }, - es_entity::ListDirection::Descending => { - #es_query_desc_call.fetch_n(op, first).await? - }, + let (entities, has_next_page) = match (direction, #(#cursor_state_scrutinee),*) { + #query_arms }; #post_hydrate_check @@ -640,21 +820,39 @@ mod tests { None }; - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { + let (entities, has_next_page) = match (direction, id.is_none()) { + (es_entity::ListDirection::Ascending, true) => { es_entity::es_query!( entity = Entity, - "SELECT id FROM entities WHERE (COALESCE(id > $2, true)) AND deleted = FALSE ORDER BY id ASC LIMIT $1", + "SELECT id FROM entities WHERE deleted = FALSE ORDER BY id ASC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Descending, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT id FROM entities WHERE deleted = FALSE ORDER BY id DESC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Ascending, false) => { + es_entity::es_query!( + entity = Entity, + "SELECT id FROM entities WHERE (id > $2) AND deleted = FALSE ORDER BY id ASC LIMIT $1", (first + 1) as i64, id as Option, ) .fetch_n(op, first) .await? }, - es_entity::ListDirection::Descending => { + (es_entity::ListDirection::Descending, false) => { es_entity::es_query!( entity = Entity, - "SELECT id FROM entities WHERE (COALESCE(id < $2, true)) AND deleted = FALSE ORDER BY id DESC LIMIT $1", + "SELECT id FROM entities WHERE (id < $2) AND deleted = FALSE ORDER BY id DESC LIMIT $1", (first + 1) as i64, id as Option, ) @@ -765,11 +963,29 @@ mod tests { (None, None) }; - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { + let (entities, has_next_page) = match (direction, id.is_none()) { + (es_entity::ListDirection::Ascending, true) => { es_entity::es_query!( entity = Entity, - "SELECT name, id FROM entities WHERE (COALESCE((name, id) > ($3, $2), $2 IS NULL)) ORDER BY name ASC, id ASC LIMIT $1", + "SELECT name, id FROM entities ORDER BY name ASC, id ASC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Descending, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT name, id FROM entities ORDER BY name DESC, id DESC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Ascending, false) => { + es_entity::es_query!( + entity = Entity, + "SELECT name, id FROM entities WHERE ((name, id) > ($3, $2)) ORDER BY name ASC, id ASC LIMIT $1", (first + 1) as i64, id as Option, name as Option, @@ -777,10 +993,10 @@ mod tests { .fetch_n(op, first) .await? }, - es_entity::ListDirection::Descending => { + (es_entity::ListDirection::Descending, false) => { es_entity::es_query!( entity = Entity, - "SELECT name, id FROM entities WHERE (COALESCE((name, id) < ($3, $2), $2 IS NULL)) ORDER BY name DESC, id DESC LIMIT $1", + "SELECT name, id FROM entities WHERE ((name, id) < ($3, $2)) ORDER BY name DESC, id DESC LIMIT $1", (first + 1) as i64, id as Option, name as Option, @@ -791,7 +1007,6 @@ mod tests { }; let end_cursor = entities.last().map(cursor_mod::EntityByNameCursor::from); - Ok(es_entity::PaginatedQueryRet { entities, has_next_page, @@ -862,11 +1077,29 @@ mod tests { (None, None) }; - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { + let (entities, has_next_page) = match (direction, id.is_some(), value.is_some()) { + (es_entity::ListDirection::Ascending, false, _) => { es_entity::es_query!( entity = Entity, - "SELECT value, id FROM entities WHERE ((value IS NOT DISTINCT FROM $3) AND COALESCE(id > $2, true) OR COALESCE(value > $3, value IS NOT NULL)) ORDER BY value ASC NULLS FIRST, id ASC LIMIT $1", + "SELECT value, id FROM entities ORDER BY value ASC NULLS FIRST, id ASC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Descending, false, _) => { + es_entity::es_query!( + entity = Entity, + "SELECT value, id FROM entities ORDER BY value DESC NULLS LAST, id DESC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Ascending, true, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT value, id FROM entities WHERE ((value, id) > ($3, $2)) ORDER BY value ASC NULLS FIRST, id ASC LIMIT $1", (first + 1) as i64, id as Option, value as Option, @@ -874,10 +1107,10 @@ mod tests { .fetch_n(op, first) .await? }, - es_entity::ListDirection::Descending => { + (es_entity::ListDirection::Descending, true, true) => { es_entity::es_query!( entity = Entity, - "SELECT value, id FROM entities WHERE ((value IS NOT DISTINCT FROM $3) AND COALESCE(id < $2, true) OR COALESCE(value < $3, $2 IS NULL OR (value IS NULL AND $3 IS NOT NULL))) ORDER BY value DESC NULLS LAST, id DESC LIMIT $1", + "SELECT value, id FROM entities WHERE ((value IS NULL OR (value, id) < ($3, $2))) ORDER BY value DESC NULLS LAST, id DESC LIMIT $1", (first + 1) as i64, id as Option, value as Option, @@ -885,10 +1118,29 @@ mod tests { .fetch_n(op, first) .await? }, + (es_entity::ListDirection::Ascending, true, false) => { + es_entity::es_query!( + entity = Entity, + "SELECT value, id FROM entities WHERE ((value IS NOT NULL OR id > $2)) ORDER BY value ASC NULLS FIRST, id ASC LIMIT $1", + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Descending, true, false) => { + es_entity::es_query!( + entity = Entity, + "SELECT value, id FROM entities WHERE ((value IS NULL AND id < $2)) ORDER BY value DESC NULLS LAST, id DESC LIMIT $1", + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, }; let end_cursor = entities.last().map(cursor_mod::EntityByValueCursor::from); - Ok(es_entity::PaginatedQueryRet { entities, has_next_page, @@ -971,8 +1223,26 @@ mod tests { (None, None) }; - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { + let (entities, has_next_page) = match (direction, id.is_none()) { + (es_entity::ListDirection::Ascending, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT value, id FROM entities ORDER BY value ASC NULLS FIRST, id ASC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Descending, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT value, id FROM entities ORDER BY value DESC NULLS LAST, id DESC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Ascending, false) => { es_entity::es_query!( entity = Entity, "SELECT value, id FROM entities WHERE ((value IS NOT DISTINCT FROM $3) AND COALESCE(id > $2, true) OR COALESCE(value > $3, value IS NOT NULL)) ORDER BY value ASC NULLS FIRST, id ASC LIMIT $1", @@ -983,7 +1253,7 @@ mod tests { .fetch_n(op, first) .await? }, - es_entity::ListDirection::Descending => { + (es_entity::ListDirection::Descending, false) => { es_entity::es_query!( entity = Entity, "SELECT value, id FROM entities WHERE ((value IS NOT DISTINCT FROM $3) AND COALESCE(id < $2, true) OR COALESCE(value < $3, $2 IS NULL OR (value IS NULL AND $3 IS NOT NULL))) ORDER BY value DESC NULLS LAST, id DESC LIMIT $1", @@ -997,7 +1267,6 @@ mod tests { }; let end_cursor = entities.last().map(cursor_mod::EntityByValueCursor::from); - Ok(es_entity::PaginatedQueryRet { entities, has_next_page, diff --git a/es-entity-macros/src/repo/list_for_filters_fn.rs b/es-entity-macros/src/repo/list_for_filters_fn.rs index a7ad7e8..fa2b9bb 100644 --- a/es-entity-macros/src/repo/list_for_filters_fn.rs +++ b/es-entity-macros/src/repo/list_for_filters_fn.rs @@ -3,7 +3,57 @@ use darling::ToTokens; use proc_macro2::{Span, TokenStream}; use quote::{TokenStreamExt, quote}; -use super::{combo_cursor::ComboCursor, list_by_fn::CursorStruct, options::*}; +use super::{ + combo_cursor::ComboCursor, + list_by_fn::{CursorStruct, assemble_select, not_deleted_predicate}, + options::*, +}; + +/// Runtime `Some`-ness state of one filter column. Each state that reaches +/// SQL gets its own static `es_query!` literal so that present filters +/// compile to sargable `col = $k` predicates instead of the non-sargable +/// `COALESCE(col = $k, $k IS NULL)` catch-all. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FilterState { + /// Filter not applied: no predicate, no parameter. + Absent, + /// Non-optional column, filter applied: `col = $k`. + Present, + /// Optional column filtering for NULL rows: `col IS NULL`, no parameter. + PresentNull, + /// Optional column filtering for a value: `col = $k`. + PresentValue, +} + +impl FilterState { + fn is_present(&self) -> bool { + !matches!(self, FilterState::Absent) + } +} + +/// Cartesian product of the per-column filter states. +fn filter_state_combos(columns: &[&Column]) -> Vec> { + columns.iter().fold(vec![vec![]], |combos, col| { + let options: &[FilterState] = if col.is_optional() { + &[ + FilterState::Absent, + FilterState::PresentValue, + FilterState::PresentNull, + ] + } else { + &[FilterState::Absent, FilterState::Present] + }; + let mut next = Vec::with_capacity(combos.len() * options.len()); + for combo in &combos { + for opt in options { + let mut combo = combo.clone(); + combo.push(*opt); + next.push(combo); + } + } + next + }) +} pub struct FiltersStruct<'a> { columns: Vec<&'a Column>, @@ -81,6 +131,18 @@ impl<'a> FiltersStruct<'a> { } } } + + /// Value-only binding for an optional column in a specialized + /// `col = $k` variant (the `apply` flag is encoded in the variant + /// itself, so only the value parameter remains). + fn filter_value_arg_tokens(column: &Column) -> TokenStream { + let col_name = column.name(); + let filter_name = syn::Ident::new(&format!("filter_{}", col_name), Span::call_site()); + let ty = column.ty(); + quote! { + #filter_name as #ty, + } + } } impl ToTokens for FiltersStruct<'_> { @@ -143,6 +205,64 @@ impl<'a> ListForFiltersFn<'a> { } } + /// Scrutinee elements (bools over the destructured filter locals) + /// identifying each filter's [`FilterState`] at runtime: one bool per + /// non-optional column (`is_some`), two per optional column (`apply`, + /// value `is_some`). + fn filter_scrutinee_elems(&self) -> Vec { + self.for_columns + .iter() + .flat_map(|c| { + let col_name = c.name(); + let filter_name = + syn::Ident::new(&format!("filter_{}", col_name), Span::call_site()); + if c.is_optional() { + let apply_name = + syn::Ident::new(&format!("apply_{}", col_name), Span::call_site()); + vec![quote! { #apply_name }, quote! { #filter_name.is_some() }] + } else { + vec![quote! { #filter_name.is_some() }] + } + }) + .collect() + } + + /// Pattern elements matching [`Self::filter_scrutinee_elems`] for one + /// column in one state. + fn filter_pattern_elems(column: &Column, state: FilterState) -> Vec { + if column.is_optional() { + match state { + FilterState::Absent => vec![quote! { false }, quote! { _ }], + FilterState::PresentValue => vec![quote! { true }, quote! { true }], + FilterState::PresentNull => vec![quote! { true }, quote! { false }], + FilterState::Present => unreachable!("optional columns split Present"), + } + } else { + match state { + FilterState::Absent => vec![quote! { false }], + FilterState::Present => vec![quote! { true }], + _ => unreachable!("non-optional columns have no NULL sub-state"), + } + } + } + + /// Whether a filter combination gets a specialized sargable query. + /// + /// Full specialization is 2^N combinations (3 per optional column) x 2 + /// cursor states x 2 directions x per sort column, so for entities with + /// many filter columns the matrix is capped: only the no-filter, + /// all-filters, and single-filter combinations are specialized and + /// everything else falls back to the catch-all COALESCE query + /// (correctness preserved, just not sargable). + fn is_specialized_combo(&self, combo: &[FilterState]) -> bool { + let n = self.for_columns.len(); + if n <= 4 { + return true; + } + let present_count = combo.iter().filter(|s| s.is_present()).count(); + present_count <= 1 || present_count == n + } + fn generate_proxy_body(&self, by_col: &Column, delete: DeleteOption) -> TokenStream { let by_col_name = by_col.name(); let delete_postfix = delete.include_deletion_fn_postfix(); @@ -330,6 +450,11 @@ impl<'a> ListForFiltersFn<'a> { .map(|col| FiltersStruct::filter_arg_tokens(col)) .collect(); + let fallback_arg_tokens = quote! { + #filter_arg_bindings + #cursor_arg_tokens + }; + let asc_query = format!( r#"SELECT {} FROM {} WHERE {}({}){} ORDER BY {} LIMIT ${}"#, select_columns, @@ -365,48 +490,135 @@ impl<'a> ListForFiltersFn<'a> { quote! {} }; - let es_query_asc_call = if let Some(prefix) = self.ignore_prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #asc_query, - #filter_arg_bindings - #cursor_arg_tokens - ) - } - } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #asc_query, - #filter_arg_bindings - #cursor_arg_tokens - ) + let make_es_query = |query: &str, args: &TokenStream| -> TokenStream { + if let Some(prefix) = self.ignore_prefix { + quote! { + es_entity::es_query!( + tbl_prefix = #prefix, + #forgettable_tbl_arg + #query, + #args + ) + } + } else { + quote! { + es_entity::es_query!( + entity = #entity, + #forgettable_tbl_arg + #query, + #args + ) + } } }; - let es_query_desc_call = if let Some(prefix) = self.ignore_prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #desc_query, - #filter_arg_bindings - #cursor_arg_tokens - ) + // Specialized variant matrix: one static query per (filter + // combination x cursor state x direction). Every present filter + // compiles to a sargable `col = $k` (or `col IS NULL`) predicate and + // the cursor predicate is either omitted (page 1) or a bare row + // comparison. + let mut asc_arms = TokenStream::new(); + let mut desc_arms = TokenStream::new(); + let mut all_combos_specialized = true; + for combo in filter_state_combos(&self.for_columns) { + if !self.is_specialized_combo(&combo) { + all_combos_specialized = false; + continue; } - } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #desc_query, - #filter_arg_bindings - #cursor_arg_tokens - ) + let filter_patterns: Vec = self + .for_columns + .iter() + .zip(combo.iter()) + .flat_map(|(col, state)| Self::filter_pattern_elems(col, *state)) + .collect(); + + let mut filter_conditions: Vec = Vec::new(); + let mut filter_args = TokenStream::new(); + let mut param_idx = 1u32; + for (col, state) in self.for_columns.iter().zip(combo.iter()) { + match state { + FilterState::Absent => {} + FilterState::Present => { + filter_conditions.push(format!("{} = ${}", col.name(), param_idx)); + param_idx += 1; + filter_args.append_all(FiltersStruct::filter_arg_tokens(col)); + } + FilterState::PresentNull => { + filter_conditions.push(format!("{} IS NULL", col.name())); + } + FilterState::PresentValue => { + filter_conditions.push(format!("{} = ${}", col.name(), param_idx)); + param_idx += 1; + filter_args.append_all(FiltersStruct::filter_value_arg_tokens(col)); + } + } + } + + for cursor_state in cursor_struct.cursor_states() { + let cursor_patterns = cursor_struct.state_pattern_elems(*cursor_state); + let pattern = quote! { (#(#filter_patterns,)* #(#cursor_patterns,)*) }; + let cursor_args = cursor_struct.cursor_arg_tokens_for_state(*cursor_state); + let args = quote! { + #filter_args + (first + 1) as i64, + #cursor_args + }; + + for ascending in [true, false] { + let mut conditions = filter_conditions.clone(); + if let Some(condition) = + cursor_struct.condition_for_state(*cursor_state, param_idx - 1, ascending) + { + conditions.push(format!("({condition})")); + } + if delete == DeleteOption::No + && let Some(not_deleted) = not_deleted_predicate(self.delete) + { + conditions.push(not_deleted); + } + let query = assemble_select( + &select_columns, + self.table_name, + &conditions, + &cursor_struct.order_by(ascending), + param_idx, + ); + let es_query_call = make_es_query(&query, &args); + if ascending { + asc_arms.append_all(quote! { + #pattern => { + #es_query_call.fetch_n(op, first).await? + }, + }); + } else { + desc_arms.append_all(quote! { + #pattern => { + #es_query_call.fetch_n(op, first).await? + }, + }); + } + } } + } + + let scrutinee_elems: Vec = self + .filter_scrutinee_elems() + .into_iter() + .chain(cursor_struct.state_scrutinee_elems()) + .collect(); + + // When every filter combination is specialized the explicit arms + // already cover the entire pattern space, so no wildcard fallback arm + // (nor its catch-all COALESCE queries) is emitted. + let (asc_fallback_arm, desc_fallback_arm) = if all_combos_specialized { + (quote! {}, quote! {}) + } else { + let asc_call = make_es_query(&asc_query, &fallback_arg_tokens); + let desc_call = make_es_query(&desc_query, &fallback_arg_tokens); + ( + quote! { _ => #asc_call.fetch_n(op, first).await?, }, + quote! { _ => #desc_call.fetch_n(op, first).await?, }, + ) }; #[cfg(feature = "instrument")] @@ -482,11 +694,13 @@ impl<'a> ListForFiltersFn<'a> { #record_fields let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { - #es_query_asc_call.fetch_n(op, first).await? + es_entity::ListDirection::Ascending => match (#(#scrutinee_elems,)*) { + #asc_arms + #asc_fallback_arm }, - es_entity::ListDirection::Descending => { - #es_query_desc_call.fetch_n(op, first).await? + es_entity::ListDirection::Descending => match (#(#scrutinee_elems,)*) { + #desc_arms + #desc_fallback_arm } }; @@ -774,29 +988,177 @@ mod tests { }; let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { - es_entity::es_query!( - entity = Order, - "SELECT id FROM orders WHERE COALESCE(customer_id = $1, $1 IS NULL) AND COALESCE(status = $2, $2 IS NULL) AND (COALESCE(id > $4, true)) ORDER BY id ASC LIMIT $3", - filter_customer_id as Option, - filter_status as Option, - (first + 1) as i64, - id as Option, - ) - .fetch_n(op, first) - .await? + es_entity::ListDirection::Ascending => match (filter_customer_id.is_some(), filter_status.is_some(), id.is_none(),) { + (false, false, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders ORDER BY id ASC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (false, false, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE (id > $2) ORDER BY id ASC LIMIT $1", + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, + (false, true, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE status = $1 ORDER BY id ASC LIMIT $2", + filter_status as Option, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (false, true, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE status = $1 AND (id > $3) ORDER BY id ASC LIMIT $2", + filter_status as Option, + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, + (true, false, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 ORDER BY id ASC LIMIT $2", + filter_customer_id as Option, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (true, false, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 AND (id > $3) ORDER BY id ASC LIMIT $2", + filter_customer_id as Option, + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, + (true, true, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 AND status = $2 ORDER BY id ASC LIMIT $3", + filter_customer_id as Option, + filter_status as Option, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (true, true, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 AND status = $2 AND (id > $4) ORDER BY id ASC LIMIT $3", + filter_customer_id as Option, + filter_status as Option, + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, }, - es_entity::ListDirection::Descending => { - es_entity::es_query!( - entity = Order, - "SELECT id FROM orders WHERE COALESCE(customer_id = $1, $1 IS NULL) AND COALESCE(status = $2, $2 IS NULL) AND (COALESCE(id < $4, true)) ORDER BY id DESC LIMIT $3", - filter_customer_id as Option, - filter_status as Option, - (first + 1) as i64, - id as Option, - ) - .fetch_n(op, first) - .await? + es_entity::ListDirection::Descending => match (filter_customer_id.is_some(), filter_status.is_some(), id.is_none(),) { + (false, false, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders ORDER BY id DESC LIMIT $1", + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (false, false, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE (id < $2) ORDER BY id DESC LIMIT $1", + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, + (false, true, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE status = $1 ORDER BY id DESC LIMIT $2", + filter_status as Option, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (false, true, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE status = $1 AND (id < $3) ORDER BY id DESC LIMIT $2", + filter_status as Option, + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, + (true, false, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 ORDER BY id DESC LIMIT $2", + filter_customer_id as Option, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (true, false, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 AND (id < $3) ORDER BY id DESC LIMIT $2", + filter_customer_id as Option, + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, + (true, true, true,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 AND status = $2 ORDER BY id DESC LIMIT $3", + filter_customer_id as Option, + filter_status as Option, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (true, true, false,) => { + es_entity::es_query!( + entity = Order, + "SELECT id FROM orders WHERE customer_id = $1 AND status = $2 AND (id < $4) ORDER BY id DESC LIMIT $3", + filter_customer_id as Option, + filter_status as Option, + (first + 1) as i64, + id as Option, + ) + .fetch_n(op, first) + .await? + }, } }; @@ -1012,10 +1374,29 @@ mod tests { let status_column = Column::new_list_for( syn::Ident::new("status", proc_macro2::Span::call_site()), syn::parse_str("String").unwrap(), - vec![id_ident], + vec![id_ident.clone()], ); - - let for_columns = vec![&workspace_id_column, &status_column]; + // Three more columns push the entity past the specialization cap so + // the catch-all COALESCE fallback (the subject of this test) is + // still emitted. + let mk_col = |name: &str| { + Column::new_list_for( + syn::Ident::new(name, proc_macro2::Span::call_site()), + syn::parse_str("String").unwrap(), + vec![id_ident.clone()], + ) + }; + let region_column = mk_col("region"); + let tier_column = mk_col("tier"); + let kind_column = mk_col("kind"); + + let for_columns = vec![ + &workspace_id_column, + &status_column, + ®ion_column, + &tier_column, + &kind_column, + ]; let by_columns = vec![&id_column]; let id_cursor = CursorStruct { @@ -1052,8 +1433,8 @@ mod tests { let token_str = tokens.to_string(); // Optional column workspace_id uses 2 params: $1 (apply bool), $2 (value) - // Non-optional column status uses 1 param: $3 - // So cursor params start at $4+ + // Non-optional columns use 1 param each: status $3, region $4, tier + // $5, kind $6. So cursor params start at $7+. assert!( token_str.contains("NOT $1 OR workspace_id IS NOT DISTINCT FROM $2"), "Expected two-param pattern for optional column, got:\n{}", @@ -1071,10 +1452,174 @@ mod tests { "Expected apply_workspace_id destructuring" ); - // LIMIT should be at $4 (3 filter params + 1) + // LIMIT should be at $7 (6 filter params + 1) + assert!( + token_str.contains("LIMIT $7"), + "Expected LIMIT at $7 (2 optional + 4 non-optional = 6 filter params)" + ); + } + + #[test] + fn list_for_filters_specializes_sargable_variants() { + let entity = Ident::new("Task", Span::call_site()); + let query_error = syn::Ident::new("TaskQueryError", Span::call_site()); + let id = syn::Ident::new("TaskId", proc_macro2::Span::call_site()); + let cursor_mod = Ident::new("cursor_mod", Span::call_site()); + + let id_column = Column::for_id(syn::parse_str("TaskId").unwrap()); + let id_ident = syn::Ident::new("id", proc_macro2::Span::call_site()); + let workspace_id_column = Column::new_list_for( + syn::Ident::new("workspace_id", proc_macro2::Span::call_site()), + syn::parse_str("Option").unwrap(), + vec![id_ident.clone()], + ); + let status_column = Column::new_list_for( + syn::Ident::new("status", proc_macro2::Span::call_site()), + syn::parse_str("String").unwrap(), + vec![id_ident], + ); + + let for_columns = vec![&workspace_id_column, &status_column]; + let by_columns = vec![&id_column]; + + let id_cursor = CursorStruct { + column: &id_column, + id: &id, + entity: &entity, + cursor_mod: &cursor_mod, + }; + + let combo_cursor = ComboCursor::new_test(&entity, vec![id_cursor]); + + let list_for_filters_fn = ListForFiltersFn { + filters_struct: FiltersStruct::new_test(&entity, for_columns.clone()), + entity: &entity, + query_error, + for_columns, + by_columns, + cursor: &combo_cursor, + delete: DeleteOption::No, + cursor_mod: cursor_mod.clone(), + table_name: "tasks", + ignore_prefix: None, + id: &id, + any_nested: false, + post_hydrate_error: None, + forgettable_table_name: None, + #[cfg(feature = "instrument")] + repo_name_snake: "test_repo".to_string(), + }; + + let mut tokens = TokenStream::new(); + list_for_filters_fn.to_tokens(&mut tokens); + let token_str = tokens.to_string(); + + let expected_queries = [ + // No filters, page 1: no WHERE at all — rides index ordering. + "SELECT id FROM tasks ORDER BY id ASC LIMIT $1", + // No filters, cursor page: bare comparison, no COALESCE. + "SELECT id FROM tasks WHERE (id > $2) ORDER BY id ASC LIMIT $1", + // Single non-optional filter. + "SELECT id FROM tasks WHERE status = $1 ORDER BY id ASC LIMIT $2", + "SELECT id FROM tasks WHERE status = $1 AND (id > $3) ORDER BY id ASC LIMIT $2", + // Optional filter on a value: sargable `col = $k`. + "SELECT id FROM tasks WHERE workspace_id = $1 ORDER BY id ASC LIMIT $2", + // Optional filter on NULL: `col IS NULL`, no parameter. + "SELECT id FROM tasks WHERE workspace_id IS NULL ORDER BY id ASC LIMIT $1", + // All filters present. + "SELECT id FROM tasks WHERE workspace_id = $1 AND status = $2 ORDER BY id ASC LIMIT $3", + "SELECT id FROM tasks WHERE workspace_id = $1 AND status = $2 AND (id > $4) ORDER BY id ASC LIMIT $3", + "SELECT id FROM tasks WHERE workspace_id IS NULL AND status = $1 ORDER BY id ASC LIMIT $2", + ]; + for query in expected_queries { + assert!( + token_str.contains(query), + "Expected specialized query `{query}` in generated code" + ); + } + + // Fully-specialized entities need no wildcard fallback arm at all — + // the explicit arms already cover the entire pattern space. + assert!( + !token_str.contains("COALESCE"), + "no COALESCE fallback should be emitted when every combination is specialized" + ); + } + + #[test] + fn list_for_filters_caps_specialization_above_four_columns() { + let entity = Ident::new("Wide", Span::call_site()); + let query_error = syn::Ident::new("WideQueryError", Span::call_site()); + let id = syn::Ident::new("WideId", proc_macro2::Span::call_site()); + let cursor_mod = Ident::new("cursor_mod", Span::call_site()); + + let id_column = Column::for_id(syn::parse_str("WideId").unwrap()); + let id_ident = syn::Ident::new("id", proc_macro2::Span::call_site()); + let mk_col = |name: &str| { + Column::new_list_for( + syn::Ident::new(name, proc_macro2::Span::call_site()), + syn::parse_str("String").unwrap(), + vec![id_ident.clone()], + ) + }; + let col_a = mk_col("a"); + let col_b = mk_col("b"); + let col_c = mk_col("c"); + let col_d = mk_col("d"); + let col_e = mk_col("e"); + + let for_columns = vec![&col_a, &col_b, &col_c, &col_d, &col_e]; + let by_columns = vec![&id_column]; + + let id_cursor = CursorStruct { + column: &id_column, + id: &id, + entity: &entity, + cursor_mod: &cursor_mod, + }; + + let combo_cursor = ComboCursor::new_test(&entity, vec![id_cursor]); + + let list_for_filters_fn = ListForFiltersFn { + filters_struct: FiltersStruct::new_test(&entity, for_columns.clone()), + entity: &entity, + query_error, + for_columns, + by_columns, + cursor: &combo_cursor, + delete: DeleteOption::No, + cursor_mod: cursor_mod.clone(), + table_name: "wides", + ignore_prefix: None, + id: &id, + any_nested: false, + post_hydrate_error: None, + forgettable_table_name: None, + #[cfg(feature = "instrument")] + repo_name_snake: "test_repo".to_string(), + }; + + let mut tokens = TokenStream::new(); + list_for_filters_fn.to_tokens(&mut tokens); + let token_str = tokens.to_string(); + + // No-filter, single-filter and all-filter combinations stay + // specialized... + assert!(token_str.contains("SELECT id FROM wides ORDER BY id ASC LIMIT $1")); + assert!(token_str.contains("SELECT id FROM wides WHERE a = $1 ORDER BY id ASC LIMIT $2")); + assert!(token_str.contains( + "SELECT id FROM wides WHERE a = $1 AND b = $2 AND c = $3 AND d = $4 AND e = $5 ORDER BY id ASC LIMIT $6" + )); + // ...but intermediate combinations (e.g. exactly two filters) fall + // back to the catch-all COALESCE query, so no specialized SQL exists + // for them. + assert!( + !token_str.contains("SELECT id FROM wides WHERE a = $1 AND b = $2 ORDER"), + "two-filter combination should not be specialized above the cap" + ); assert!( - token_str.contains("LIMIT $4"), - "Expected LIMIT at $4 (2 optional + 1 non-optional = 3 filter params)" + token_str.contains("COALESCE(a = $1, $1 IS NULL)"), + "COALESCE fallback must remain for uncapped combinations" ); } } diff --git a/es-entity-macros/src/repo/list_for_fn.rs b/es-entity-macros/src/repo/list_for_fn.rs index 914aab3..675de4c 100644 --- a/es-entity-macros/src/repo/list_for_fn.rs +++ b/es-entity-macros/src/repo/list_for_fn.rs @@ -2,7 +2,10 @@ use darling::ToTokens; use proc_macro2::{Span, TokenStream}; use quote::{TokenStreamExt, quote}; -use super::{list_by_fn::CursorStruct, options::*}; +use super::{ + list_by_fn::{CursorStruct, assemble_select, not_deleted_predicate}, + options::*, +}; pub struct ListForFn<'a> { ignore_prefix: Option<&'a syn::LitStr>, @@ -74,7 +77,6 @@ impl ToTokens for ListForFn<'_> { let destructure_tokens = self.cursor().destructure_tokens(); let select_columns = cursor.select_columns(Some(for_column_name)); - let arg_tokens = cursor.query_arg_tokens(); for delete in [DeleteOption::No, DeleteOption::Soft] { let fn_name = syn::Ident::new( @@ -101,34 +103,6 @@ impl ToTokens for ListForFn<'_> { } else { "=" }; - let asc_query = format!( - r#"SELECT {} FROM {} WHERE (({} {} $1) AND ({})){} ORDER BY {} LIMIT $2"#, - select_columns, - self.table_name, - for_column_name, - filter_op, - cursor.condition(1, true), - if delete == DeleteOption::No { - self.delete.not_deleted_condition() - } else { - "" - }, - cursor.order_by(true) - ); - let desc_query = format!( - r#"SELECT {} FROM {} WHERE (({} {} $1) AND ({})){} ORDER BY {} LIMIT $2"#, - select_columns, - self.table_name, - for_column_name, - filter_op, - cursor.condition(1, false), - if delete == DeleteOption::No { - self.delete.not_deleted_condition() - } else { - "" - }, - cursor.order_by(false) - ); let forgettable_tbl_arg = if let Some(tbl) = self.forgettable_table_name { quote! { forgettable_tbl = #tbl, } @@ -136,49 +110,69 @@ impl ToTokens for ListForFn<'_> { quote! {} }; - let es_query_asc_call = if let Some(prefix) = self.ignore_prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #asc_query, - #filter_arg_name as &#for_column_type, - #arg_tokens - ) - } - } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #asc_query, - #filter_arg_name as &#for_column_type, - #arg_tokens - ) + let make_es_query = |query: &str, args: &TokenStream| -> TokenStream { + if let Some(prefix) = self.ignore_prefix { + quote! { + es_entity::es_query!( + tbl_prefix = #prefix, + #forgettable_tbl_arg + #query, + #args + ) + } + } else { + quote! { + es_entity::es_query!( + entity = #entity, + #forgettable_tbl_arg + #query, + #args + ) + } } }; - let es_query_desc_call = if let Some(prefix) = self.ignore_prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #desc_query, - #filter_arg_name as &#for_column_type, - #arg_tokens - ) - } - } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #desc_query, + let mut query_arms = TokenStream::new(); + for state in cursor.cursor_states() { + for ascending in [true, false] { + let mut conditions: Vec = + vec![format!("({for_column_name} {filter_op} $1)")]; + if let Some(condition) = cursor.condition_for_state(*state, 1, ascending) { + conditions.push(format!("({condition})")); + } + if delete == DeleteOption::No + && let Some(not_deleted) = not_deleted_predicate(self.delete) + { + conditions.push(not_deleted); + } + let query = assemble_select( + &select_columns, + self.table_name, + &conditions, + &cursor.order_by(ascending), + 2, + ); + let cursor_args = cursor.cursor_arg_tokens_for_state(*state); + let args = quote! { #filter_arg_name as &#for_column_type, - #arg_tokens - ) + (first + 1) as i64, + #cursor_args + }; + let es_query_call = make_es_query(&query, &args); + let direction_pattern = if ascending { + quote! { es_entity::ListDirection::Ascending } + } else { + quote! { es_entity::ListDirection::Descending } + }; + let state_pattern = cursor.state_pattern_elems(*state); + query_arms.append_all(quote! { + (#direction_pattern, #(#state_pattern),*) => { + #es_query_call.fetch_n(op, first).await? + }, + }); } - }; + } + let cursor_state_scrutinee = cursor.state_scrutinee_elems(); #[cfg(feature = "instrument")] let ( @@ -270,13 +264,8 @@ impl ToTokens for ListForFn<'_> { #destructure_tokens #record_fields - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { - #es_query_asc_call.fetch_n(op, first).await? - }, - es_entity::ListDirection::Descending => { - #es_query_desc_call.fetch_n(op, first).await? - } + let (entities, has_next_page) = match (direction, #(#cursor_state_scrutinee),*) { + #query_arms }; #post_hydrate_check @@ -369,11 +358,31 @@ mod tests { } else { None }; - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { + let (entities, has_next_page) = match (direction, id.is_none()) { + (es_entity::ListDirection::Ascending, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT customer_id, id FROM entities WHERE (customer_id = $1) ORDER BY id ASC LIMIT $2", + filter_customer_id as &Uuid, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Descending, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT customer_id, id FROM entities WHERE (customer_id = $1) ORDER BY id DESC LIMIT $2", + filter_customer_id as &Uuid, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Ascending, false) => { es_entity::es_query!( entity = Entity, - "SELECT customer_id, id FROM entities WHERE ((customer_id = $1) AND (COALESCE(id > $3, true))) ORDER BY id ASC LIMIT $2", + "SELECT customer_id, id FROM entities WHERE (customer_id = $1) AND (id > $3) ORDER BY id ASC LIMIT $2", filter_customer_id as &Uuid, (first + 1) as i64, id as Option, @@ -381,25 +390,25 @@ mod tests { .fetch_n(op, first) .await? }, - es_entity::ListDirection::Descending => { + (es_entity::ListDirection::Descending, false) => { es_entity::es_query!( entity = Entity, - "SELECT customer_id, id FROM entities WHERE ((customer_id = $1) AND (COALESCE(id < $3, true))) ORDER BY id DESC LIMIT $2", + "SELECT customer_id, id FROM entities WHERE (customer_id = $1) AND (id < $3) ORDER BY id DESC LIMIT $2", filter_customer_id as &Uuid, (first + 1) as i64, id as Option, ) .fetch_n(op, first) .await? - } + }, }; - let end_cursor = entities.last().map(cursor_mod::EntityByIdCursor::from); - Ok(es_entity::PaginatedQueryRet { - entities, - has_next_page, - end_cursor, - }) + let end_cursor = entities.last().map(cursor_mod::EntityByIdCursor::from); + Ok(es_entity::PaginatedQueryRet { + entities, + has_next_page, + end_cursor, + }) }.await; __result @@ -468,11 +477,31 @@ mod tests { } else { (None, None) }; - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => { + let (entities, has_next_page) = match (direction, id.is_none()) { + (es_entity::ListDirection::Ascending, true) => { es_entity::es_query!( entity = Entity, - "SELECT email, id FROM entities WHERE ((email = $1) AND (COALESCE((email, id) > ($4, $3), $3 IS NULL))) ORDER BY email ASC, id ASC LIMIT $2", + "SELECT email, id FROM entities WHERE (email = $1) ORDER BY email ASC, id ASC LIMIT $2", + filter_email as &str, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Descending, true) => { + es_entity::es_query!( + entity = Entity, + "SELECT email, id FROM entities WHERE (email = $1) ORDER BY email DESC, id DESC LIMIT $2", + filter_email as &str, + (first + 1) as i64, + ) + .fetch_n(op, first) + .await? + }, + (es_entity::ListDirection::Ascending, false) => { + es_entity::es_query!( + entity = Entity, + "SELECT email, id FROM entities WHERE (email = $1) AND ((email, id) > ($4, $3)) ORDER BY email ASC, id ASC LIMIT $2", filter_email as &str, (first + 1) as i64, id as Option, @@ -481,10 +510,10 @@ mod tests { .fetch_n(op, first) .await? }, - es_entity::ListDirection::Descending => { + (es_entity::ListDirection::Descending, false) => { es_entity::es_query!( entity = Entity, - "SELECT email, id FROM entities WHERE ((email = $1) AND (COALESCE((email, id) < ($4, $3), $3 IS NULL))) ORDER BY email DESC, id DESC LIMIT $2", + "SELECT email, id FROM entities WHERE (email = $1) AND ((email, id) < ($4, $3)) ORDER BY email DESC, id DESC LIMIT $2", filter_email as &str, (first + 1) as i64, id as Option, @@ -492,7 +521,7 @@ mod tests { ) .fetch_n(op, first) .await? - } + }, }; let end_cursor = entities.last().map(cursor_mod::EntityByEmailCursor::from); diff --git a/migrations/20260724000000_sargable_list_test.sql b/migrations/20260724000000_sargable_list_test.sql new file mode 100644 index 0000000..88cac9c --- /dev/null +++ b/migrations/20260724000000_sargable_list_test.sql @@ -0,0 +1,21 @@ +CREATE TABLE transfers ( + id UUID PRIMARY KEY, + account_id UUID NOT NULL, + status VARCHAR NOT NULL, + reference VARCHAR DEFAULT NULL, + score INT DEFAULT NULL, + created_at TIMESTAMPTZ NOT NULL +); +CREATE INDEX idx_transfers_account_created_id ON transfers (account_id, created_at DESC, id DESC); +CREATE INDEX idx_transfers_status ON transfers (status); +CREATE INDEX idx_transfers_score_id ON transfers (score, id); + +CREATE TABLE transfer_events ( + id UUID NOT NULL REFERENCES transfers(id), + sequence INT NOT NULL, + event_type VARCHAR NOT NULL, + event JSONB NOT NULL, + context JSONB DEFAULT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + UNIQUE(id, sequence) +); diff --git a/tests/entities/mod.rs b/tests/entities/mod.rs index 448444d..3623ab0 100644 --- a/tests/entities/mod.rs +++ b/tests/entities/mod.rs @@ -2,4 +2,5 @@ pub mod customer; pub mod order; pub mod profile; pub mod task; +pub mod transfer; pub mod user; diff --git a/tests/entities/transfer.rs b/tests/entities/transfer.rs new file mode 100644 index 0000000..3886c33 --- /dev/null +++ b/tests/entities/transfer.rs @@ -0,0 +1,100 @@ +#![allow(dead_code)] + +use derive_builder::Builder; +use serde::{Deserialize, Serialize}; + +use es_entity::*; + +es_entity::entity_id! { TransferId } +es_entity::entity_id! { AccountId } + +/// Mirrors the shape of lana's `Disbursal` repo: two non-optional `list_for` +/// filter columns (`account_id`, `status`), one optional filter column +/// (`reference`), and a nullable sort column (`score`) for NULL-cursor edge +/// cases. +#[derive(EsEvent, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +#[es_event(id = "TransferId")] +pub enum TransferEvent { + Initialized { + id: TransferId, + account_id: AccountId, + status: String, + reference: Option, + score: Option, + }, +} + +#[derive(EsEntity, Builder)] +#[builder(pattern = "owned", build_fn(error = "EntityHydrationError"))] +pub struct Transfer { + pub id: TransferId, + pub account_id: AccountId, + pub status: String, + #[builder(default)] + pub reference: Option, + #[builder(default)] + pub score: Option, + + events: EntityEvents, +} + +impl TryFromEvents for Transfer { + fn try_from_events(events: EntityEvents) -> Result { + let mut builder = TransferBuilder::default(); + for event in events.iter_all() { + match event { + TransferEvent::Initialized { + id, + account_id, + status, + reference, + score, + } => { + builder = builder + .id(*id) + .account_id(*account_id) + .status(status.clone()) + .reference(reference.clone()) + .score(*score); + } + } + } + builder.events(events).build() + } +} + +#[derive(Debug, Builder)] +pub struct NewTransfer { + #[builder(setter(into))] + pub id: TransferId, + #[builder(setter(into))] + pub account_id: AccountId, + #[builder(setter(into))] + pub status: String, + #[builder(setter(into, strip_option), default)] + pub reference: Option, + #[builder(default)] + pub score: Option, +} + +impl NewTransfer { + pub fn builder() -> NewTransferBuilder { + NewTransferBuilder::default() + } +} + +impl IntoEvents for NewTransfer { + fn into_events(self) -> EntityEvents { + EntityEvents::init( + self.id, + [TransferEvent::Initialized { + id: self.id, + account_id: self.account_id, + status: self.status, + reference: self.reference, + score: self.score, + }], + ) + } +} diff --git a/tests/sargable_list_queries.rs b/tests/sargable_list_queries.rs new file mode 100644 index 0000000..9d32c4e --- /dev/null +++ b/tests/sargable_list_queries.rs @@ -0,0 +1,331 @@ +mod entities; +mod helpers; + +use chrono::{DateTime, Utc}; +use sqlx::PgPool; + +use entities::transfer::*; +use es_entity::*; + +/// Repo shape mirroring lana's `Disbursals`: two non-optional filter columns +/// plus one optional filter column, sorted `by(created_at)`, plus a nullable +/// `score` sort column for NULL-cursor edge cases. +#[derive(EsRepo, Debug)] +#[es_repo( + entity = "Transfer", + columns( + account_id(ty = "AccountId", list_for(by(created_at))), + status(ty = "String", list_for(by(created_at))), + reference(ty = "Option", list_for), + score(ty = "Option", list_by) + ) +)] +pub struct Transfers { + pool: PgPool, +} + +impl Transfers { + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[derive(Debug, Clone)] +struct Row { + id: uuid::Uuid, + account_id: uuid::Uuid, + status: String, + reference: Option, + score: Option, + created_at: DateTime, +} + +async fn seed_transfers( + repo: &Transfers, + specs: &[(uuid::Uuid, &str, Option<&str>, Option)], +) -> anyhow::Result<()> { + for (account_id, status, reference, score) in specs { + let mut new = NewTransfer::builder() + .id(TransferId::new()) + .account_id(AccountId::from(*account_id)) + .status(*status) + .score(*score) + .build() + .unwrap(); + new.reference = reference.map(|r| r.to_string()); + repo.create(new).await?; + } + Ok(()) +} + +async fn ground_truth(pool: &PgPool, account_ids: &[uuid::Uuid]) -> anyhow::Result> { + let rows = sqlx::query_as::< + _, + ( + uuid::Uuid, + uuid::Uuid, + String, + Option, + Option, + DateTime, + ), + >( + "SELECT id, account_id, status, reference, score, created_at FROM transfers WHERE account_id = ANY($1)", + ) + .bind(account_ids) + .fetch_all(pool) + .await?; + Ok(rows + .into_iter() + .map( + |(id, account_id, status, reference, score, created_at)| Row { + id, + account_id, + status, + reference, + score, + created_at, + }, + ) + .collect()) +} + +/// Paginating `list_for_filters` through every filter combination x sort x +/// direction must return exactly the same rows in exactly the same order as +/// an in-Rust reference implementation. This is the correctness harness for +/// the specialized query matrix — it exercises the proxy dispatch (dedicated +/// `list_for_*` / `list_by_*` paths), the specialized catch-all variants, +/// and the cursor/no-cursor split on every page transition. +#[tokio::test] +async fn list_for_filters_matches_reference_for_all_combos() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let transfers = Transfers::new(pool.clone()); + + let account_ids: Vec = (0..3).map(|_| uuid::Uuid::from(AccountId::new())).collect(); + let mut specs = Vec::new(); + for i in 0..40 { + let account_id = account_ids[i % 3]; + let status = if i % 2 == 0 { "active" } else { "settled" }; + let reference = match i % 3 { + 0 => None, + 1 => Some("ref-a"), + _ => Some("ref-b"), + }; + specs.push((account_id, status, reference, None)); + } + seed_transfers(&transfers, &specs).await?; + let truth = ground_truth(&pool, &account_ids).await?; + // The table is shared with other tests (and previous runs): scenarios + // without an account filter legitimately return foreign rows. Ordering + // is total and deterministic, so filtering the returned stream down to + // this test's rows preserves everything there is to check. + let truth_ids: std::collections::HashSet<_> = truth.iter().map(|r| r.id).collect(); + + let account_filters: [Option; 2] = [None, Some(account_ids[0])]; + let status_filters: [Option<&str>; 2] = [None, Some("active")]; + let reference_filters: [Option>; 3] = [None, Some(None), Some(Some("ref-a"))]; + let sorts = [TransferSortBy::Id, TransferSortBy::CreatedAt]; + let directions = [ListDirection::Ascending, ListDirection::Descending]; + + for account_filter in account_filters { + for status_filter in status_filters { + for reference_filter in reference_filters { + for by in sorts { + for direction in directions { + let expected = reference_order( + &truth, + account_filter, + status_filter, + reference_filter, + by, + direction, + ); + + let mut actual = Vec::new(); + let mut after: Option = None; + loop { + let ret = transfers + .list_for_filters( + TransferFilters { + account_id: account_filter.map(AccountId::from), + status: status_filter.map(|s| s.to_string()), + reference: reference_filter + .map(|r| r.map(|s| s.to_string())), + }, + Sort { by, direction }, + PaginatedQueryArgs { first: 7, after }, + ) + .await?; + actual.extend(ret.entities.iter().map(|t| uuid::Uuid::from(t.id))); + if !ret.has_next_page { + break; + } + after = ret.end_cursor; + assert!(after.is_some(), "has_next_page without end_cursor"); + } + actual.retain(|id| truth_ids.contains(id)); + + assert_eq!( + actual, expected, + "mismatch for account={account_filter:?} status={status_filter:?} \ + reference={reference_filter:?} by={by:?} direction={direction:?}" + ); + } + } + } + } + } + + Ok(()) +} + +fn reference_order( + rows: &[Row], + account_filter: Option, + status_filter: Option<&str>, + reference_filter: Option>, + by: TransferSortBy, + direction: ListDirection, +) -> Vec { + let mut rows: Vec<&Row> = rows + .iter() + .filter(|r| account_filter.is_none_or(|a| r.account_id == a)) + .filter(|r| status_filter.is_none_or(|s| r.status == s)) + .filter(|r| match reference_filter { + None => true, + Some(None) => r.reference.is_none(), + Some(Some(v)) => r.reference.as_deref() == Some(v), + }) + .collect(); + rows.sort_by(|a, b| match by { + TransferSortBy::Id => a.id.cmp(&b.id), + _ => (a.created_at, a.id).cmp(&(b.created_at, b.id)), + }); + if matches!(direction, ListDirection::Descending) { + rows.reverse(); + } + rows.into_iter().map(|r| r.id).collect() +} + +/// `list_by_score` over a nullable sort column must paginate correctly +/// through NULL and non-NULL cursor values in both directions: +/// ASC sorts NULLs FIRST, DESC sorts NULLs LAST. +#[tokio::test] +async fn list_by_score_paginates_through_nulls() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let transfers = Transfers::new(pool.clone()); + + let account_id = uuid::Uuid::from(AccountId::new()); + let scores = [ + None, + Some(5), + None, + Some(3), + Some(7), + None, + Some(1), + Some(3), + ]; + let specs: Vec<_> = scores + .iter() + .map(|score| (account_id, "score_test", None, *score)) + .collect(); + seed_transfers(&transfers, &specs).await?; + let truth = ground_truth(&pool, &[account_id]).await?; + // `list_by_score` is unfiltered over a shared table — retain only this + // test's rows from the (totally ordered) returned stream. + let truth_ids: std::collections::HashSet<_> = truth.iter().map(|r| r.id).collect(); + + for direction in [ListDirection::Ascending, ListDirection::Descending] { + // Reference: ASC -> NULLs first (id asc), then values by (score, id) + // asc. DESC -> values by (score, id) desc, then NULLs (id desc). + let mut null_rows: Vec<_> = truth.iter().filter(|r| r.score.is_none()).collect(); + let mut value_rows: Vec<_> = truth.iter().filter(|r| r.score.is_some()).collect(); + null_rows.sort_by_key(|r| r.id); + value_rows.sort_by_key(|r| (r.score, r.id)); + let expected: Vec = match direction { + ListDirection::Ascending => null_rows + .into_iter() + .chain(value_rows) + .map(|r| r.id) + .collect(), + ListDirection::Descending => value_rows + .into_iter() + .rev() + .chain(null_rows.into_iter().rev()) + .map(|r| r.id) + .collect(), + }; + + // A page size smaller than the 8 seeded rows forces pagination even + // on a pristine database: ASC crosses the NULL → value boundary and + // DESC the value → NULL boundary, so the specialized `After` and + // `AfterNull` cursor variants both execute in each direction. + let mut actual = Vec::new(); + let mut after: Option = None; + let mut pages = 0; + loop { + let ret = transfers + .list_by_score(PaginatedQueryArgs { first: 3, after }, direction) + .await?; + pages += 1; + actual.extend(ret.entities.iter().map(|t| uuid::Uuid::from(t.id))); + if !ret.has_next_page { + break; + } + after = ret.end_cursor; + assert!(after.is_some(), "has_next_page without end_cursor"); + } + actual.retain(|id| truth_ids.contains(id)); + + assert!( + pages >= 3, + "pagination must span multiple pages to exercise the cursor \ + variants, got {pages} page(s) for direction={direction:?}" + ); + assert_eq!(actual, expected, "mismatch for direction={direction:?}"); + } + + Ok(()) +} + +/// The dedicated single-filter path (what lana's hot +/// `creditFacility { disbursals }` resolver should dispatch to) filters and +/// paginates correctly on its own. +#[tokio::test] +async fn list_for_account_id_by_created_at_paginates() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let transfers = Transfers::new(pool.clone()); + + let account_ids: Vec = (0..2).map(|_| uuid::Uuid::from(AccountId::new())).collect(); + let specs: Vec<_> = (0..10) + .map(|i| (account_ids[i % 2], "dedicated", None, None)) + .collect(); + seed_transfers(&transfers, &specs).await?; + let truth = ground_truth(&pool, &[account_ids[0]]).await?; + + let mut expected: Vec<_> = truth.iter().collect(); + expected.sort_by_key(|r| (r.created_at, r.id)); + let expected: Vec<_> = expected.into_iter().rev().map(|r| r.id).collect(); + + let mut actual = Vec::new(); + let mut after: Option = None; + loop { + let ret = transfers + .list_for_account_id_by_created_at( + AccountId::from(account_ids[0]), + PaginatedQueryArgs { first: 2, after }, + ListDirection::Descending, + ) + .await?; + actual.extend(ret.entities.iter().map(|t| uuid::Uuid::from(t.id))); + if !ret.has_next_page { + break; + } + after = ret.end_cursor; + } + + assert_eq!(actual, expected); + assert_eq!(actual.len(), 5); + Ok(()) +}