densa_arrow/paged/source.rs
1//! What a paged model reads (a source of rows, a page at a time, in the order asked for), how
2//! it pages, and why rows can be missing.
3
4use std::error::Error;
5use std::fmt;
6use std::ops::Range;
7
8use arrow_array::RecordBatch;
9use densa_grid::SortOrder;
10
11/// What a [`PageSource`] fails with: any error, sent across threads.
12pub type PageError = Box<dyn Error + Send + Sync>;
13
14/// Rows a page holds unless told otherwise: several screens.
15const PAGE_ROWS: usize = 256;
16/// Pages kept unless told otherwise.
17const PAGES_KEPT: usize = 64;
18
19/// A column rows are sorted by, and which way.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct SortKey {
22 column: String,
23 order: SortOrder,
24}
25
26impl SortKey {
27 /// Rows by `column` (its name) in `order`.
28 pub fn new(column: impl Into<String>, order: SortOrder) -> Self {
29 Self {
30 column: column.into(),
31 order,
32 }
33 }
34
35 /// The column's name.
36 pub fn column(&self) -> &str {
37 &self.column
38 }
39
40 /// Which way.
41 pub fn order(&self) -> SortOrder {
42 self.order
43 }
44
45 /// Whether the largest come first.
46 pub fn is_descending(&self) -> bool {
47 self.order == SortOrder::Descending
48 }
49}
50
51/// What a [`PageSource`] is asked for: its rows, in an order (an `ORDER BY`).
52#[derive(Debug, Clone, Default, PartialEq)]
53pub struct PageQuery {
54 sort: Vec<SortKey>,
55}
56
57impl PageQuery {
58 /// The source's rows in its own order.
59 pub fn new() -> Self {
60 Self::default()
61 }
62
63 /// The query with its rows sorted by `key` too, after the keys it has.
64 #[must_use]
65 pub fn sorted_by(mut self, key: SortKey) -> Self {
66 self.sort.push(key);
67 self
68 }
69
70 /// The keys rows are sorted by, the first first; none for the source's own order.
71 pub fn sort(&self) -> &[SortKey] {
72 &self.sort
73 }
74
75 /// The order as an SQL `ORDER BY` clause, for a source that pages a query with `LIMIT` and
76 /// `OFFSET`: the sort keys, then the `keys` not among them, ascending. Give as `keys` the
77 /// columns that tell rows apart (an id): rows that tie in every sort key then keep one
78 /// order from page to page, so none shows twice or goes missing. Names are quoted as
79 /// standard SQL quotes them (`"time"`, and `"a ""b"""` for a column named `a "b"`), each
80 /// once. Empty with no sort keys and no `keys`, which leaves the order to the engine: one
81 /// that reads in parallel may change it from one page to the next.
82 ///
83 /// ```
84 /// use densa_arrow::{PageQuery, SortKey};
85 /// use densa_grid::SortOrder;
86 ///
87 /// let query = PageQuery::new().sorted_by(SortKey::new("time", SortOrder::Descending));
88 /// let order = query.order_by_sql(&["id"]);
89 /// assert_eq!(order, r#"ORDER BY "time" DESC, "id""#);
90 /// let sql = format!("SELECT * FROM trips {order} LIMIT 256 OFFSET 512");
91 /// ```
92 pub fn order_by_sql<S: AsRef<str>>(&self, keys: &[S]) -> String {
93 let sorted = self
94 .sort
95 .iter()
96 .map(|key| (key.column(), key.is_descending()));
97 let then = keys.iter().map(|key| (key.as_ref(), false));
98 let mut named: Vec<&str> = Vec::new();
99 let mut sql = String::new();
100 for (column, descending) in sorted.chain(then) {
101 if named.contains(&column) {
102 continue;
103 }
104 sql.push_str(if named.is_empty() { "ORDER BY " } else { ", " });
105 named.push(column);
106 sql.push('"');
107 sql.push_str(&column.replace('"', "\"\""));
108 sql.push('"');
109 if descending {
110 sql.push_str(" DESC");
111 }
112 }
113 sql
114 }
115}
116
117/// The rows of a table too large to hold, a page at a time: what a
118/// [`PagedModel`](super::PagedModel) reads. Its calls (but [`can_sort`](Self::can_sort)) run on
119/// a thread of the model's own, so a call may take as long as it needs: a query to a database
120/// or a query engine. A call is not interrupted; the next waits for it. Rows are numbered from
121/// 0 in the order the query gives them.
122///
123/// A call that fails, or panics (with panics unwinding, the default), is reported by the model
124/// ([`PagedModel::error`](super::PagedModel::error)), and the next calls go on.
125pub trait PageSource: Send + 'static {
126 /// How many rows `query` gives.
127 ///
128 /// # Errors
129 /// Any the source meets: the model shows no rows ([`FetchError::Count`]) until
130 /// [`retry`](super::PagedModel::retry) or [`refresh`](super::PagedModel::refresh) counts
131 /// them again.
132 fn row_count(&mut self, query: &PageQuery) -> Result<u64, PageError>;
133
134 /// `limit` rows from row `offset` of what `query` gives (fewer only at the end), in as many
135 /// batches as they come in, each with the model's columns: their data types, in order (names
136 /// are not compared). The model asks for pages side by side in one call; when such a call
137 /// fails, it asks for each of its pages alone.
138 ///
139 /// # Errors
140 /// Any the source meets: those rows stay empty ([`FetchError::Rows`]) until
141 /// [`retry`](super::PagedModel::retry) or [`refresh`](super::PagedModel::refresh). Fewer
142 /// rows than asked for before the end say the data changed since it was counted
143 /// ([`FetchError::Short`]).
144 fn page(
145 &mut self,
146 query: &PageQuery,
147 offset: u64,
148 limit: usize,
149 ) -> Result<Vec<RecordBatch>, PageError>;
150
151 /// Whether the source can give its rows sorted by `column`: a click on its title sorts only
152 /// then. Asked once for each column when the model is made, on the thread making it; every
153 /// column can by default.
154 fn can_sort(&self, column: &str) -> bool {
155 let _ = column;
156 true
157 }
158}
159
160/// How a [`PagedModel`](super::PagedModel) pages: the rows of a page and how many pages it
161/// keeps.
162#[derive(Debug, Clone)]
163#[non_exhaustive]
164pub struct PageOptions {
165 /// Rows a page holds: 256 unless set, at least 1.
166 pub page_rows: usize,
167 /// Pages kept, 64 unless set, at least 1: those farthest from the rows shown go first
168 /// (fetched again if shown again). Pages of the rows shown are kept even past it.
169 pub pages_kept: usize,
170}
171
172impl Default for PageOptions {
173 fn default() -> Self {
174 Self {
175 page_rows: PAGE_ROWS,
176 pages_kept: PAGES_KEPT,
177 }
178 }
179}
180
181impl PageOptions {
182 /// The options with pages of `rows` rows.
183 #[must_use]
184 pub fn page_rows(mut self, rows: usize) -> Self {
185 self.page_rows = rows;
186 self
187 }
188
189 /// The options keeping `pages` pages.
190 #[must_use]
191 pub fn pages_kept(mut self, pages: usize) -> Self {
192 self.pages_kept = pages;
193 self
194 }
195}
196
197/// Why a [`PagedModel`](super::PagedModel) has rows it cannot show: what
198/// [`PagedModel::error`](super::PagedModel::error) gives.
199#[derive(Debug, Clone, PartialEq, Eq)]
200#[non_exhaustive]
201pub enum FetchError {
202 /// The source failed to count the rows (`message` says why): no rows are shown until
203 /// [`retry`](super::PagedModel::retry) or [`refresh`](super::PagedModel::refresh) counts
204 /// them.
205 Count {
206 /// The source's error, or what its panic said.
207 message: String,
208 },
209 /// The source failed to give `rows` (`message` says why): they stay empty until
210 /// [`retry`](super::PagedModel::retry) or [`refresh`](super::PagedModel::refresh).
211 Rows {
212 /// The rows asked for, those of one page.
213 rows: Range<u64>,
214 /// The source's error, or what its panic said.
215 message: String,
216 },
217 /// The source gave only `got` of `rows`, which its count says are there: the data changed
218 /// since it was counted. They stay empty until [`refresh`](super::PagedModel::refresh)
219 /// counts again.
220 Short {
221 /// The rows asked for, those of one page.
222 rows: Range<u64>,
223 /// How many came.
224 got: u64,
225 },
226 /// The thread fetching pages stopped (`message` says why: the model's `wake` panicked):
227 /// nothing more arrives, so make the model again.
228 Stopped {
229 /// What stopped it.
230 message: String,
231 },
232}
233
234impl fmt::Display for FetchError {
235 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
236 match self {
237 Self::Count { message } => write!(f, "the rows could not be counted: {message}"),
238 Self::Rows { rows, message } => write!(
239 f,
240 "rows {} to {} could not be read: {message}",
241 rows.start,
242 rows.end.saturating_sub(1)
243 ),
244 Self::Short { rows, got } => write!(
245 f,
246 "rows {} to {} came back short ({got} of {}): the data changed since it was \
247 counted; refresh to count it again",
248 rows.start,
249 rows.end.saturating_sub(1),
250 rows.end.saturating_sub(rows.start)
251 ),
252 Self::Stopped { message } => write!(
253 f,
254 "the thread fetching pages stopped ({message}): make the model again"
255 ),
256 }
257 }
258}
259
260impl Error for FetchError {}