fraiseql_wire/client/query_builder/mod.rs
1//! Query builder API
2//!
3//! Generic query builder that supports automatic JSON deserialization to target types.
4//!
5//! **IMPORTANT**: Type T is **consumer-side only**.
6//!
7//! Type T does NOT affect:
8//! - SQL generation (always `SELECT data FROM {entity}`)
9//! - Filtering (`where_sql`, `where_rust`, `order_by`)
10//! - Wire protocol (identical for all T)
11//!
12//! Type T ONLY affects:
13//! - Consumer-side deserialization at `poll_next()`
14//! - Error messages (type name included)
15
16use crate::client::FraiseClient;
17#[allow(unused_imports)] // Reason: used only in doc links for `# Errors` sections
18use crate::error::WireError;
19use crate::stream::QueryStream;
20use crate::Result;
21use serde::de::DeserializeOwned;
22use serde_json::Value;
23use std::marker::PhantomData;
24
25/// Type alias for a Rust-side predicate function
26type RustPredicate = Box<dyn Fn(&Value) -> bool + Send>;
27
28/// Generic query builder
29///
30/// The type parameter T controls consumer-side deserialization only.
31/// Default type T = `serde_json::Value` for backward compatibility.
32///
33/// # Examples
34///
35/// Type-safe query (recommended):
36/// ```no_run
37/// // Requires: live Postgres connection via FraiseClient.
38/// use serde::Deserialize;
39///
40/// #[derive(Deserialize)]
41/// struct Project {
42/// id: String,
43/// name: String,
44/// }
45/// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
46/// let stream = client.query::<Project>("projects")
47/// .where_sql("status='active'")
48/// .execute()
49/// .await?;
50/// # Ok(())
51/// # }
52/// ```
53///
54/// Raw JSON query (debugging, forward compatibility):
55/// ```no_run
56/// // Requires: live Postgres connection via FraiseClient.
57/// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
58/// let stream = client.query::<serde_json::Value>("projects")
59/// .execute()
60/// .await?;
61/// # Ok(())
62/// # }
63/// ```
64#[must_use = "call .execute() to run the query"]
65pub struct QueryBuilder<T: DeserializeOwned + Unpin + 'static = serde_json::Value> {
66 client: FraiseClient,
67 entity: String,
68 sql_predicates: Vec<String>,
69 rust_predicate: Option<RustPredicate>,
70 order_by: Option<String>,
71 limit: Option<usize>,
72 offset: Option<usize>,
73 chunk_size: usize,
74 max_memory: Option<usize>,
75 soft_limit_warn_threshold: Option<f32>, // Percentage (0.0-1.0) at which to warn
76 soft_limit_fail_threshold: Option<f32>, // Percentage (0.0-1.0) at which to error
77 enable_adaptive_chunking: bool,
78 adaptive_min_chunk_size: Option<usize>,
79 adaptive_max_chunk_size: Option<usize>,
80 custom_select: Option<String>, // Optional custom SELECT clause for SQL projection
81 _phantom: PhantomData<T>,
82}
83
84impl<T: DeserializeOwned + Unpin + 'static> QueryBuilder<T> {
85 /// Create new query builder
86 pub(crate) fn new(client: FraiseClient, entity: impl Into<String>) -> Self {
87 Self {
88 client,
89 entity: entity.into(),
90 sql_predicates: Vec::new(),
91 rust_predicate: None,
92 order_by: None,
93 limit: None,
94 offset: None,
95 chunk_size: 256,
96 max_memory: None,
97 soft_limit_warn_threshold: None,
98 soft_limit_fail_threshold: None,
99 // Off by default: this preserves the historical *effective* behaviour
100 // (the option was previously dropped and forced off at the execute
101 // boundary). Now that the option is honoured (audit L-wire-builder),
102 // opt in explicitly with `.adaptive_chunking(true)` — fixed-size
103 // chunking stays the zero-overhead default.
104 enable_adaptive_chunking: false,
105 adaptive_min_chunk_size: None,
106 adaptive_max_chunk_size: None,
107 custom_select: None,
108 _phantom: PhantomData,
109 }
110 }
111
112 /// Add SQL WHERE clause predicate
113 ///
114 /// Type T does NOT affect SQL generation.
115 /// Multiple predicates are AND'ed together.
116 pub fn where_sql(mut self, predicate: impl Into<String>) -> Self {
117 self.sql_predicates.push(predicate.into());
118 self
119 }
120
121 /// Add Rust-side predicate
122 ///
123 /// Type T does NOT affect filtering.
124 /// Applied after SQL filtering, runs on streamed JSON values.
125 /// Predicates receive &`serde_json::Value` regardless of T.
126 pub fn where_rust<F>(mut self, predicate: F) -> Self
127 where
128 F: Fn(&Value) -> bool + Send + 'static,
129 {
130 self.rust_predicate = Some(Box::new(predicate));
131 self
132 }
133
134 /// Set ORDER BY clause
135 ///
136 /// Type T does NOT affect ordering.
137 pub fn order_by(mut self, order: impl Into<String>) -> Self {
138 self.order_by = Some(order.into());
139 self
140 }
141
142 /// Set a custom SELECT clause for SQL projection optimization
143 ///
144 /// When provided, this replaces the default `SELECT data` with a projection SQL
145 /// that filters fields at the database level, reducing network payload.
146 ///
147 /// The projection SQL will be wrapped as `SELECT {projection_sql} as data` to maintain
148 /// the hard invariant of a single `data` column.
149 ///
150 /// This feature enables architectural consistency with PostgreSQL optimization
151 /// and prepares for future performance improvements.
152 ///
153 /// # Arguments
154 ///
155 /// * `projection_sql` - PostgreSQL expression, typically from `jsonb_build_object()`
156 ///
157 /// # Example
158 ///
159 /// ```no_run
160 /// // Requires: live Postgres connection via FraiseClient.
161 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
162 /// # use serde::Deserialize;
163 /// # #[derive(Deserialize)] struct Project { id: String, name: String }
164 /// let stream = client
165 /// .query::<Project>("projects")
166 /// .select_projection("jsonb_build_object('id', data->>'id', 'name', data->>'name')")
167 /// .execute()
168 /// .await?;
169 /// # Ok(())
170 /// # }
171 /// ```
172 ///
173 /// # Backward Compatibility
174 ///
175 /// If not specified, defaults to `SELECT data` (original behavior).
176 pub fn select_projection(mut self, projection_sql: impl Into<String>) -> Self {
177 self.custom_select = Some(projection_sql.into());
178 self
179 }
180
181 /// Set LIMIT clause to restrict result set size
182 ///
183 /// # Example
184 ///
185 /// ```no_run
186 /// // Requires: live Postgres connection via FraiseClient.
187 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
188 /// # use serde::Deserialize;
189 /// # #[derive(Deserialize)] struct Project { id: String }
190 /// let stream = client.query::<Project>("projects")
191 /// .limit(10)
192 /// .execute()
193 /// .await?;
194 /// # Ok(())
195 /// # }
196 /// ```
197 pub const fn limit(mut self, count: usize) -> Self {
198 self.limit = Some(count);
199 self
200 }
201
202 /// Set OFFSET clause to skip first N rows
203 ///
204 /// # Example
205 ///
206 /// ```no_run
207 /// // Requires: live Postgres connection via FraiseClient.
208 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
209 /// # use serde::Deserialize;
210 /// # #[derive(Deserialize)] struct Project { id: String }
211 /// let stream = client.query::<Project>("projects")
212 /// .limit(10)
213 /// .offset(20) // Skip first 20, return next 10
214 /// .execute()
215 /// .await?;
216 /// # Ok(())
217 /// # }
218 /// ```
219 pub const fn offset(mut self, count: usize) -> Self {
220 self.offset = Some(count);
221 self
222 }
223
224 /// Set chunk size (default: 256)
225 pub const fn chunk_size(mut self, size: usize) -> Self {
226 self.chunk_size = size;
227 self
228 }
229
230 /// Set maximum memory limit for buffered items (default: unbounded)
231 ///
232 /// When the estimated memory usage of buffered items exceeds this limit,
233 /// the stream will return `WireError::MemoryLimitExceeded` instead of additional items.
234 ///
235 /// Memory is estimated as: `items_buffered * 2048 bytes` (conservative for typical JSON).
236 ///
237 /// By default, `max_memory()` is None (unbounded), maintaining backward compatibility.
238 /// Only set if you need hard memory bounds.
239 ///
240 /// # Example
241 ///
242 /// ```no_run
243 /// // Requires: live Postgres connection via FraiseClient.
244 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
245 /// # use serde::Deserialize;
246 /// # #[derive(Deserialize)] struct Project { id: String }
247 /// let stream = client
248 /// .query::<Project>("projects")
249 /// .max_memory(500_000_000) // 500 MB limit
250 /// .execute()
251 /// .await?;
252 /// # Ok(())
253 /// # }
254 /// ```
255 ///
256 /// # Interpretation
257 ///
258 /// If memory limit is exceeded:
259 /// - It indicates the consumer is too slow relative to data arrival
260 /// - The error is terminal (non-retriable) — retrying won't help
261 /// - Consider: increasing consumer throughput, reducing `chunk_size`, or removing limit
262 pub const fn max_memory(mut self, bytes: usize) -> Self {
263 self.max_memory = Some(bytes);
264 self
265 }
266
267 /// Set soft memory limit thresholds for progressive degradation
268 ///
269 /// Allows warning at a threshold before hitting hard limit.
270 /// Only applies if `max_memory()` is also set.
271 ///
272 /// # Parameters
273 ///
274 /// - `warn_threshold`: Percentage (0.0-1.0) at which to emit a warning
275 /// - `fail_threshold`: Percentage (0.0-1.0) at which to return error (must be > `warn_threshold`)
276 ///
277 /// # Example
278 ///
279 /// ```no_run
280 /// // Requires: live Postgres connection via FraiseClient.
281 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
282 /// # use serde::Deserialize;
283 /// # #[derive(Deserialize)] struct Project { id: String }
284 /// let stream = client
285 /// .query::<Project>("projects")
286 /// .max_memory(500_000_000) // 500 MB hard limit
287 /// .memory_soft_limits(0.80, 1.0) // Warn at 80%, error at 100%
288 /// .execute()
289 /// .await?;
290 /// # Ok(())
291 /// # }
292 /// ```
293 ///
294 /// If only hard limit needed, skip this and just use `max_memory()`.
295 pub fn memory_soft_limits(mut self, warn_threshold: f32, fail_threshold: f32) -> Self {
296 // Validate thresholds
297 let warn = warn_threshold.clamp(0.0, 1.0);
298 let fail = fail_threshold.clamp(0.0, 1.0);
299
300 if warn < fail {
301 self.soft_limit_warn_threshold = Some(warn);
302 self.soft_limit_fail_threshold = Some(fail);
303 }
304 self
305 }
306
307 /// Enable or disable adaptive chunk sizing (default: disabled)
308 ///
309 /// Adaptive chunking automatically adjusts `chunk_size` based on channel occupancy:
310 /// - High occupancy (>80%): Decreases chunk size to reduce producer pressure
311 /// - Low occupancy (<20%): Increases chunk size to optimize batching efficiency
312 ///
313 /// Disabled by default — fixed-size chunking is the zero-overhead path. Enable
314 /// it for self-tuning batch sizes when consumer throughput varies.
315 ///
316 /// # Example
317 ///
318 /// ```no_run
319 /// // Requires: live Postgres connection via FraiseClient.
320 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
321 /// # use serde::Deserialize;
322 /// # #[derive(Deserialize)] struct Project { id: String }
323 /// let stream = client
324 /// .query::<Project>("projects")
325 /// .adaptive_chunking(true) // Enable adaptive tuning
326 /// .execute()
327 /// .await?;
328 /// # Ok(())
329 /// # }
330 /// ```
331 pub const fn adaptive_chunking(mut self, enabled: bool) -> Self {
332 self.enable_adaptive_chunking = enabled;
333 self
334 }
335
336 /// Override minimum chunk size for adaptive tuning (default: 16)
337 ///
338 /// Adaptive chunking will never decrease chunk size below this value.
339 /// Useful if you need minimum batching for performance.
340 ///
341 /// Only applies if adaptive chunking is enabled.
342 ///
343 /// # Example
344 ///
345 /// ```no_run
346 /// // Requires: live Postgres connection via FraiseClient.
347 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
348 /// # use serde::Deserialize;
349 /// # #[derive(Deserialize)] struct Project { id: String }
350 /// let stream = client
351 /// .query::<Project>("projects")
352 /// .adaptive_chunking(true)
353 /// .adaptive_min_size(32) // Don't go below 32 items per batch
354 /// .execute()
355 /// .await?;
356 /// # Ok(())
357 /// # }
358 /// ```
359 pub const fn adaptive_min_size(mut self, size: usize) -> Self {
360 self.adaptive_min_chunk_size = Some(size);
361 self
362 }
363
364 /// Override maximum chunk size for adaptive tuning (default: 1024)
365 ///
366 /// Adaptive chunking will never increase chunk size above this value.
367 /// Useful if you need memory bounds or latency guarantees.
368 ///
369 /// Only applies if adaptive chunking is enabled.
370 ///
371 /// # Example
372 ///
373 /// ```no_run
374 /// // Requires: live Postgres connection via FraiseClient.
375 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
376 /// # use serde::Deserialize;
377 /// # #[derive(Deserialize)] struct Project { id: String }
378 /// let stream = client
379 /// .query::<Project>("projects")
380 /// .adaptive_chunking(true)
381 /// .adaptive_max_size(512) // Cap at 512 items per batch
382 /// .execute()
383 /// .await?;
384 /// # Ok(())
385 /// # }
386 /// ```
387 pub const fn adaptive_max_size(mut self, size: usize) -> Self {
388 self.adaptive_max_chunk_size = Some(size);
389 self
390 }
391
392 /// Execute query and return typed stream
393 ///
394 /// Type T ONLY affects consumer-side deserialization at `poll_next()`.
395 /// SQL, filtering, ordering, and wire protocol are identical regardless of T.
396 ///
397 /// The returned stream supports pause/resume/stats for advanced stream control.
398 ///
399 /// # Examples
400 ///
401 /// With type-safe deserialization:
402 /// ```no_run
403 /// // Requires: live Postgres connection via FraiseClient.
404 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
405 /// # use serde::Deserialize;
406 /// # #[derive(Deserialize)] struct Project { id: String }
407 /// # use futures::stream::StreamExt;
408 /// let mut stream = client.query::<Project>("projects").execute().await?;
409 /// while let Some(result) = stream.next().await {
410 /// let project: Project = result?;
411 /// }
412 /// # Ok(())
413 /// # }
414 /// ```
415 ///
416 /// With raw JSON (escape hatch):
417 /// ```no_run
418 /// // Requires: live Postgres connection via FraiseClient.
419 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
420 /// # use futures::stream::StreamExt;
421 /// let mut stream = client.query::<serde_json::Value>("projects").execute().await?;
422 /// while let Some(result) = stream.next().await {
423 /// let json: serde_json::Value = result?;
424 /// }
425 /// # Ok(())
426 /// # }
427 /// ```
428 ///
429 /// With stream control:
430 /// ```no_run
431 /// // Requires: live Postgres connection via FraiseClient.
432 /// # async fn example(client: fraiseql_wire::FraiseClient) -> fraiseql_wire::Result<()> {
433 /// let mut stream = client.query::<serde_json::Value>("projects").execute().await?;
434 /// stream.pause().await?; // Pause the stream
435 /// let stats = stream.stats(); // Get statistics
436 /// stream.resume().await?; // Resume the stream
437 /// # Ok(())
438 /// # }
439 /// ```
440 ///
441 /// # Errors
442 ///
443 /// Returns [`WireError`] if SQL generation fails or the underlying streaming query
444 /// cannot be started on the connection.
445 pub async fn execute(self) -> Result<QueryStream<T>> {
446 let sql = self.build_sql()?;
447 tracing::debug!("executing query: {}", sql);
448
449 // Record query submission metrics
450 crate::metrics::counters::query_submitted(
451 &self.entity,
452 !self.sql_predicates.is_empty(),
453 self.rust_predicate.is_some(),
454 self.order_by.is_some(),
455 );
456
457 let stream = self
458 .client
459 .execute_query(
460 &sql,
461 self.chunk_size,
462 self.max_memory,
463 self.soft_limit_warn_threshold,
464 self.soft_limit_fail_threshold,
465 self.enable_adaptive_chunking,
466 self.adaptive_min_chunk_size,
467 self.adaptive_max_chunk_size,
468 )
469 .await?;
470
471 // Create QueryStream with optional Rust predicate
472 Ok(QueryStream::new(stream, self.rust_predicate))
473 }
474
475 /// Build SQL query
476 fn build_sql(&self) -> Result<String> {
477 // Use custom SELECT clause if provided, otherwise default to "SELECT data"
478 let select_clause = if let Some(ref projection) = self.custom_select {
479 format!("SELECT {} as data", projection)
480 } else {
481 "SELECT data".to_string()
482 };
483
484 let mut sql = format!("{} FROM {}", select_clause, self.entity);
485
486 if !self.sql_predicates.is_empty() {
487 sql.push_str(" WHERE ");
488 sql.push_str(&self.sql_predicates.join(" AND "));
489 }
490
491 if let Some(ref order) = self.order_by {
492 sql.push_str(" ORDER BY ");
493 sql.push_str(order);
494 }
495
496 if let Some(limit) = self.limit {
497 sql.push_str(&format!(" LIMIT {}", limit));
498 }
499
500 if let Some(offset) = self.offset {
501 sql.push_str(&format!(" OFFSET {}", offset));
502 }
503
504 Ok(sql)
505 }
506}
507
508#[cfg(test)]
509mod tests;