Skip to main content

perspective_client/
table.rs

1// ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
2// ┃ ██████ ██████ ██████       █      █      █      █      █ █▄  ▀███ █       ┃
3// ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█  ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄  ▀█ █ ▀▀▀▀▀ ┃
4// ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄   █ ▄▄▄▄▄ ┃
5// ┃ █      ██████ █  ▀█▄       █ ██████      █      ███▌▐███ ███████▄ █       ┃
6// ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫
7// ┃ Copyright (c) 2017, the Perspective Authors.                              ┃
8// ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃
9// ┃ This file is part of the Perspective library, distributed under the terms ┃
10// ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
11// ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
12
13use std::collections::HashMap;
14use std::fmt::Display;
15
16use indexmap::IndexMap;
17use serde::{Deserialize, Serialize};
18use ts_rs::TS;
19
20use crate::assert_table_api;
21use crate::client::{Client, Features};
22use crate::config::{Expressions, ViewConfigUpdate};
23use crate::proto::make_table_req::MakeTableOptions;
24use crate::proto::make_table_req::make_table_options::MakeTableType;
25use crate::proto::request::ClientReq;
26use crate::proto::response::ClientResp;
27use crate::proto::*;
28use crate::table_data::UpdateData;
29use crate::utils::*;
30use crate::view::{View, ViewSource};
31
32pub type Schema = HashMap<String, ColumnType>;
33
34/// The format to interpret data preovided to [`Client::table`].
35///
36/// When serialized, these values are `"csv"`, `"json"`, `"columns"`, `"arrow"`
37/// and `"ndjson"`.
38#[derive(Clone, Copy, Debug, Serialize, Deserialize, TS)]
39pub enum TableReadFormat {
40    #[serde(rename = "csv")]
41    Csv,
42
43    #[serde(rename = "json")]
44    JsonString,
45
46    #[serde(rename = "columns")]
47    ColumnsString,
48
49    #[serde(rename = "arrow")]
50    Arrow,
51
52    #[serde(rename = "ndjson")]
53    Ndjson,
54}
55
56impl TableReadFormat {
57    pub fn parse(value: Option<String>) -> Result<Option<Self>, String> {
58        Ok(match value.as_deref() {
59            Some("csv") => Some(TableReadFormat::Csv),
60            Some("json") => Some(TableReadFormat::JsonString),
61            Some("columns") => Some(TableReadFormat::ColumnsString),
62            Some("arrow") => Some(TableReadFormat::Arrow),
63            Some("ndjson") => Some(TableReadFormat::Ndjson),
64            None => None,
65            Some(x) => return Err(format!("Unknown format \"{x}\"")),
66        })
67    }
68}
69
70/// Options which impact the behavior of [`Client::table`], as well as
71/// subsequent calls to [`Table::update`].
72#[derive(Clone, Debug, Default, Serialize, Deserialize, TS)]
73pub struct TableInitOptions {
74    #[serde(default)]
75    #[ts(optional)]
76    pub name: Option<String>,
77
78    #[serde(default)]
79    #[ts(optional)]
80    pub format: Option<TableReadFormat>,
81
82    /// This [`Table`] should use the column named by the `index` parameter as
83    /// the `index`, which causes [`Table::update`] and [`Client::table`] input
84    /// to either insert or update existing rows based on `index` column
85    /// value equality.
86    #[serde(default)]
87    #[ts(optional)]
88    pub index: Option<String>,
89
90    /// This [`Table`] should be limited to `limit` rows, after which the
91    /// _earliest_ rows will be overwritten (where _earliest_ is defined as
92    /// relative to insertion order).
93    #[serde(default)]
94    #[ts(optional)]
95    pub limit: Option<u32>,
96
97    /// Back this [`Table`]'s canonical data with the on-disk storage backend
98    /// instead of memory. On native targets this is a memory-mapped file; on
99    /// WASM it is OPFS (Worker only). Defaults to in-memory.
100    #[serde(default)]
101    #[ts(optional)]
102    pub page_to_disk: Option<bool>,
103
104    /// How Arrow `LIST` and JSON `Array` columns are ingested. `zip` (the
105    /// default) and `cartesian` expand a row into one row per list element,
106    /// and are incompatible with `index`, as the rows of an expansion
107    /// repeat their index. `stringify` encodes each list as a JSON array in
108    /// a single string column instead.
109    #[serde(default)]
110    #[ts(optional)]
111    pub list_flatten: Option<crate::proto::ListFlatten>,
112
113    /// The columns of a [`Table`] made from a [`View`], in place of the ones
114    /// inferred from it.
115    #[serde(default)]
116    #[ts(optional)]
117    pub schema: Option<IndexMap<String, ColumnType>>,
118}
119
120impl TableInitOptions {
121    pub fn set_name<D: Display>(&mut self, name: D) {
122        self.name = Some(format!("{name}"))
123    }
124}
125
126impl TryFrom<TableOptions> for MakeTableOptions {
127    type Error = ClientError;
128
129    fn try_from(value: TableOptions) -> Result<Self, Self::Error> {
130        let page_to_disk = value.page_to_disk;
131        let list_flatten = value.list_flatten.map(|x| x as i32);
132        let view_schema = value.schema.clone().map(|schema| crate::proto::Schema {
133            schema: schema
134                .into_iter()
135                .map(|(name, r#type)| crate::proto::schema::KeyTypePair {
136                    name,
137                    r#type: r#type as i32,
138                })
139                .collect(),
140        });
141
142        Ok(MakeTableOptions {
143            page_to_disk,
144            list_flatten,
145            view_schema,
146            make_table_type: match value {
147                TableOptions {
148                    index: Some(_),
149                    limit: Some(_),
150                    ..
151                } => Err(ClientError::BadTableOptions)?,
152                TableOptions {
153                    index: Some(index), ..
154                } => Some(MakeTableType::MakeIndexTable(index)),
155                TableOptions {
156                    limit: Some(limit), ..
157                } => Some(MakeTableType::MakeLimitTable(limit)),
158                _ => None,
159            },
160        })
161    }
162}
163
164#[derive(Clone, Debug)]
165pub(crate) struct TableOptions {
166    pub index: Option<String>,
167    pub limit: Option<u32>,
168    pub page_to_disk: Option<bool>,
169    pub list_flatten: Option<crate::proto::ListFlatten>,
170    pub schema: Option<IndexMap<String, ColumnType>>,
171}
172
173/// The source [`View`] of a replica [`Table`] built by [`Client::table`],
174/// with the subscription tokens to release when the replica is deleted.
175#[derive(Clone)]
176pub(crate) struct ViewBinding {
177    pub view: View,
178    pub update_token: u32,
179    pub remove_token: Option<u32>,
180}
181
182impl From<TableInitOptions> for TableOptions {
183    fn from(value: TableInitOptions) -> Self {
184        TableOptions {
185            index: value.index,
186            limit: value.limit,
187            page_to_disk: value.page_to_disk,
188            list_flatten: value.list_flatten,
189            schema: value.schema,
190        }
191    }
192}
193
194/// Options for [`Client::join`].
195#[derive(Clone, Debug, Default, Serialize, Deserialize, TS)]
196pub struct JoinOptions {
197    #[serde(default)]
198    #[ts(optional)]
199    pub join_type: Option<crate::proto::JoinType>,
200
201    #[serde(default)]
202    #[ts(optional)]
203    pub name: Option<String>,
204
205    #[serde(default)]
206    #[ts(optional)]
207    pub right_on: Option<String>,
208}
209
210/// Options for [`Table::delete`].
211#[derive(Clone, Debug, Default, Deserialize, TS)]
212pub struct DeleteOptions {
213    pub lazy: bool,
214}
215
216/// Options for [`Table::update`].
217#[derive(Clone, Debug, Default, Deserialize, Serialize, TS)]
218pub struct UpdateOptions {
219    pub port_id: Option<u32>,
220    pub format: Option<TableReadFormat>,
221}
222
223/// Result of a call to [`Table::validate_expressions`], containing a schema
224/// for valid expressions and error messages for invalid ones.
225#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
226pub struct ExprValidationResult {
227    pub expression_schema: Schema,
228    pub errors: HashMap<String, ExpressionError>,
229    pub expression_alias: HashMap<String, String>,
230}
231
232/// The successful result of [`Table::describe`]: what a [`View`] built from the
233/// described config would report, WITHOUT building one.
234#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
235pub struct Description {
236    /// The type of every expression, before aggregation.
237    pub expression_schema: Schema,
238
239    /// Equal to [`View::schema`] of a [`View`] built from the same config.
240    pub view_schema: Schema,
241}
242
243/// Why [`Table::describe`] rejected a config.
244#[derive(Clone, Debug, PartialEq)]
245pub enum DescribeError {
246    /// At least one expression failed to compile.
247    Expressions {
248        expression_schema: Schema,
249        errors: HashMap<String, ExpressionError>,
250    },
251
252    /// Every expression compiled but the rest of the config is invalid.
253    Config(String),
254}
255
256impl Display for DescribeError {
257    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
258        match self {
259            Self::Expressions { errors, .. } => {
260                let mut names = errors.keys().collect::<Vec<_>>();
261                names.sort();
262                write!(f, "Invalid expressions: ")?;
263                for (i, name) in names.iter().enumerate() {
264                    if i > 0 {
265                        write!(f, "; ")?;
266                    }
267
268                    write!(f, "{}: {}", name, errors[*name].error_message)?;
269                }
270
271                Ok(())
272            },
273            Self::Config(msg) => write!(f, "{}", msg),
274        }
275    }
276}
277
278impl std::error::Error for DescribeError {}
279
280/// The wire/serde shape of a [`Table::describe`] verdict, as exchanged with
281/// JavaScript and Python (both client results and virtual-server handler
282/// returns): exactly one of the three arms, discriminated by its keys.
283#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
284#[serde(untagged)]
285pub enum DescribeVerdict {
286    Ok {
287        expression_schema: Schema,
288        view_schema: Schema,
289    },
290    Expressions {
291        #[serde(default)]
292        expression_schema: Schema,
293        expression_errors: HashMap<String, ExpressionError>,
294    },
295    Config {
296        config_error: String,
297    },
298}
299
300impl From<Result<Description, DescribeError>> for DescribeVerdict {
301    fn from(value: Result<Description, DescribeError>) -> Self {
302        match value {
303            Ok(d) => Self::Ok {
304                expression_schema: d.expression_schema,
305                view_schema: d.view_schema,
306            },
307            Err(DescribeError::Expressions {
308                expression_schema,
309                errors,
310            }) => Self::Expressions {
311                expression_schema,
312                expression_errors: errors,
313            },
314            Err(DescribeError::Config(config_error)) => Self::Config { config_error },
315        }
316    }
317}
318
319impl TryFrom<DescribeVerdict> for Result<Description, DescribeError> {
320    type Error = ClientError;
321
322    /// Fails only for the unrepresentable `Expressions` arm with no errors.
323    fn try_from(value: DescribeVerdict) -> Result<Self, ClientError> {
324        Ok(match value {
325            DescribeVerdict::Ok {
326                expression_schema,
327                view_schema,
328            } => Ok(Description {
329                expression_schema,
330                view_schema,
331            }),
332            DescribeVerdict::Expressions {
333                expression_schema,
334                expression_errors,
335            } => {
336                if expression_errors.is_empty() {
337                    return Err(ClientError::Unknown(
338                        "Describe verdict has an empty `expression_errors`".to_string(),
339                    ));
340                }
341
342                Err(DescribeError::Expressions {
343                    expression_schema,
344                    errors: expression_errors,
345                })
346            },
347            DescribeVerdict::Config { config_error } => Err(DescribeError::Config(config_error)),
348        })
349    }
350}
351
352fn decode_schema(schema: HashMap<String, i32>) -> ClientResult<Schema> {
353    schema
354        .into_iter()
355        .map(|(name, ty)| {
356            ColumnType::try_from(ty)
357                .map(|ty| (name, ty))
358                .map_err(|e| ClientError::Unknown(e.to_string()))
359        })
360        .collect()
361}
362
363/// [`Table`] is Perspective's columnar data frame, analogous to a Pandas/Polars
364/// `DataFrame` or Apache Arrow, supporting append & in-place updates, removal
365/// by index, and update notifications.
366///
367/// A [`Table`] contains columns, each of which have a unique name, are strongly
368/// and consistently typed, and contains rows of data conforming to the column's
369/// type. Each column in a [`Table`] must have the same number of rows, though
370/// not every row must contain data; null-values are used to indicate missing
371/// values in the dataset. The schema of a [`Table`] is _immutable after
372/// creation_, which means the column names and data types cannot be changed
373/// after the [`Table`] has been created. Columns cannot be added or deleted
374/// after creation either, but a [`View`] can be used to select an arbitrary set
375/// of columns from the [`Table`].
376#[derive(Clone)]
377pub struct Table {
378    name: String,
379    client: Client,
380    options: TableOptions,
381    pub(crate) view_binding: Option<ViewBinding>,
382}
383
384assert_table_api!(Table);
385
386impl PartialEq for Table {
387    fn eq(&self, other: &Self) -> bool {
388        self.name == other.name && self.client == other.client
389    }
390}
391
392impl Table {
393    pub(crate) fn new(name: String, client: Client, options: TableOptions) -> Self {
394        Table {
395            name,
396            client,
397            options,
398            view_binding: None,
399        }
400    }
401
402    fn client_message(&self, req: ClientReq) -> Request {
403        Request {
404            msg_id: self.client.gen_id(),
405            entity_id: self.name.clone(),
406            client_req: Some(req),
407        }
408    }
409
410    /// Get a copy of the [`Client`] this [`Table`] came from.
411    pub fn get_client(&self) -> Client {
412        self.client.clone()
413    }
414
415    /// Get a metadata dictionary of the `perspective_server::Server`'s
416    /// features, which is (currently) implementation specific, but there is
417    /// only one implementation.
418    pub async fn get_features(&self) -> ClientResult<Features> {
419        self.client.get_features().await
420    }
421
422    /// Returns the name of the index column for the table.
423    ///
424    /// # Examples
425    ///
426    /// ```no_run
427    /// # use perspective_client::{Client, TableData, TableInitOptions, UpdateData};
428    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
429    /// # let client: Client = todo!();
430    /// let options = TableInitOptions {
431    ///     index: Some("x".to_string()),
432    ///     ..TableInitOptions::default()
433    /// };
434    /// let data = TableData::Update(UpdateData::Csv("x,y\n1,2\n3,4".into()));
435    /// let table = client.table(data, options).await?;
436    /// let index = table.get_index();
437    /// # Ok(()) }
438    /// ```
439    pub fn get_index(&self) -> Option<String> {
440        self.options.index.as_ref().map(|index| index.to_owned())
441    }
442
443    /// Returns the user-specified row limit for this table.
444    pub fn get_limit(&self) -> Option<u32> {
445        self.options.limit.as_ref().map(|limit| *limit)
446    }
447
448    /// Returns the user-specified name for this table, or the auto-generated
449    /// name if a name was not specified when the table was created.
450    pub fn get_name(&self) -> &str {
451        self.name.as_str()
452    }
453
454    /// Removes all the rows in the [`Table`], but preserves everything else
455    /// including the schema, index, and any callbacks or registered
456    /// [`View`] instances.
457    ///
458    /// Calling [`Table::clear`], like [`Table::update`] and [`Table::remove`],
459    /// will trigger an update event to any registered listeners via
460    /// [`View::on_update`].
461    pub async fn clear(&self) -> ClientResult<()> {
462        self.replace(UpdateData::JsonRows("[]".to_owned())).await
463    }
464
465    /// Delete this [`Table`] and cleans up associated resources.
466    ///
467    /// [`Table`]s do not stop consuming resources or processing updates when
468    /// they are garbage collected in their host language - you must call
469    /// this method to reclaim these.
470    ///
471    /// # Arguments
472    ///
473    /// - `options` An options dictionary.
474    ///     - `lazy` Whether to delete this [`Table`] _lazily_. When false (the
475    ///       default), the delete will occur immediately, assuming it has no
476    ///       [`View`] instances registered to it (which must be deleted first,
477    ///       otherwise this method will throw an error). When true, the
478    ///       [`Table`] will only be marked for deltion once its [`View`]
479    ///       dependency count reaches 0.
480    ///
481    /// # Examples
482    ///
483    /// ```no_run
484    /// # use perspective_client::{Client, DeleteOptions, TableData, TableInitOptions, UpdateData};
485    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
486    /// # let client: Client = todo!();
487    /// let opts = TableInitOptions::default();
488    /// let data = TableData::Update(UpdateData::Csv("x,y\n1,2\n3,4".into()));
489    /// let table = client.table(data, opts).await?;
490    ///
491    /// // ...
492    ///
493    /// table.delete(DeleteOptions::default()).await?;
494    /// # Ok(()) }
495    /// ```
496    pub async fn delete(&self, options: DeleteOptions) -> ClientResult<()> {
497        if let Some(binding) = &self.view_binding {
498            binding.view.remove_update(binding.update_token).await?;
499            if let Some(token) = binding.remove_token {
500                binding.view.remove_remove(token).await?;
501            }
502        }
503
504        let msg = self.client_message(ClientReq::TableDeleteReq(TableDeleteReq {
505            is_immediate: !options.lazy,
506        }));
507
508        match self.client.oneshot(&msg).await? {
509            ClientResp::TableDeleteResp(_) => Ok(()),
510            resp => Err(resp.into()),
511        }
512    }
513
514    /// Returns the column names of this [`Table`] in "natural" order (the
515    /// ordering implied by the input format).
516    ///  
517    /// # Examples
518    ///
519    /// ```no_run
520    /// # use perspective_client::Table;
521    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
522    /// # let table: Table = todo!();
523    /// let columns = table.columns().await?;
524    /// # Ok(()) }
525    /// ```
526    pub async fn columns(&self) -> ClientResult<Vec<String>> {
527        let msg = self.client_message(ClientReq::TableSchemaReq(TableSchemaReq {}));
528        match self.client.oneshot(&msg).await? {
529            ClientResp::TableSchemaResp(TableSchemaResp { schema }) => Ok(schema
530                .map(|x| x.schema.into_iter().map(|x| x.name.to_owned()).collect())
531                .unwrap()),
532            resp => Err(resp.into()),
533        }
534    }
535
536    /// Returns the number of rows in a [`Table`].
537    pub async fn size(&self) -> ClientResult<usize> {
538        let msg = self.client_message(ClientReq::TableSizeReq(TableSizeReq {}));
539        match self.client.oneshot(&msg).await? {
540            ClientResp::TableSizeResp(TableSizeResp { size }) => Ok(size as usize),
541            resp => Err(resp.into()),
542        }
543    }
544
545    /// Returns a table's [`Schema`], a mapping of column names to column types.
546    ///
547    /// The mapping of a [`Table`]'s column names to data types is referred to
548    /// as a [`Schema`]. Each column has a unique name and a data type, one
549    /// of:
550    ///
551    /// - `"boolean"` - A boolean type
552    /// - `"date"` - A timesonze-agnostic date type (month/day/year)
553    /// - `"datetime"` - A millisecond-precision datetime type in the UTC
554    ///   timezone
555    /// - `"float"` - A 64 bit float
556    /// - `"integer"` - A signed 32 bit integer (the integer type supported by
557    ///   JavaScript)
558    /// - `"string"` - A [`String`] data type (encoded internally as a
559    ///   _dictionary_)
560    ///
561    /// Note that all [`Table`] columns are _nullable_, regardless of the data
562    /// type.
563    pub async fn schema(&self) -> ClientResult<Schema> {
564        let msg = self.client_message(ClientReq::TableSchemaReq(TableSchemaReq {}));
565        match self.client.oneshot(&msg).await? {
566            ClientResp::TableSchemaResp(TableSchemaResp { schema }) => Ok(schema
567                .map(|x| {
568                    x.schema
569                        .into_iter()
570                        .map(|x| (x.name, ColumnType::try_from(x.r#type).unwrap()))
571                        .collect()
572                })
573                .unwrap()),
574            resp => Err(resp.into()),
575        }
576    }
577
578    /// Create a unique channel ID on this [`Table`], which allows
579    /// `View::on_update` callback calls to be associated with the
580    /// `Table::update` which caused them.
581    pub async fn make_port(&self) -> ClientResult<i32> {
582        let msg = self.client_message(ClientReq::TableMakePortReq(TableMakePortReq {}));
583        match self.client.oneshot(&msg).await? {
584            ClientResp::TableMakePortResp(TableMakePortResp { port_id }) => Ok(port_id as i32),
585            _ => Err(ClientError::Unknown("make_port".to_string())),
586        }
587    }
588
589    /// Register a callback which is called exactly once, when this [`Table`] is
590    /// deleted with the [`Table::delete`] method.
591    ///
592    /// [`Table::on_delete`] resolves when the subscription message is sent, not
593    /// when the _delete_ event occurs.
594    pub async fn on_delete(
595        &self,
596        on_delete: Box<dyn Fn() + Send + Sync + 'static>,
597    ) -> ClientResult<u32> {
598        let callback = move |resp: Response| match resp.client_resp {
599            Some(ClientResp::TableOnDeleteResp(_)) => {
600                on_delete();
601                Ok(())
602            },
603            resp => Err(resp.into()),
604        };
605
606        let msg = self.client_message(ClientReq::TableOnDeleteReq(TableOnDeleteReq {}));
607        self.client.subscribe_once(&msg, Box::new(callback)).await?;
608        Ok(msg.msg_id)
609    }
610
611    /// Removes a listener with a given ID, as returned by a previous call to
612    /// [`Table::on_delete`].
613    pub async fn remove_delete(&self, callback_id: u32) -> ClientResult<()> {
614        let msg = self.client_message(ClientReq::TableRemoveDeleteReq(TableRemoveDeleteReq {
615            id: callback_id,
616        }));
617
618        match self.client.oneshot(&msg).await? {
619            ClientResp::TableRemoveDeleteResp(_) => Ok(()),
620            resp => Err(resp.into()),
621        }
622    }
623
624    /// Removes rows from this [`Table`] with the `index` column values
625    /// supplied.
626    ///
627    /// # Arguments
628    ///
629    /// - `indices` - A list of `index` column values for rows that should be
630    ///   removed.
631    ///
632    /// # Examples
633    ///
634    /// ```no_run
635    /// # use perspective_client::{Table, UpdateData};
636    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
637    /// # let table: Table = todo!();
638    /// table
639    ///     .remove(UpdateData::Csv("index\n1\n2\n3".into()))
640    ///     .await?;
641    /// # Ok(()) }
642    /// ```
643    pub async fn remove(&self, input: UpdateData) -> ClientResult<()> {
644        let msg = self.client_message(ClientReq::TableRemoveReq(TableRemoveReq {
645            data: Some(input.into()),
646        }));
647
648        match self.client.oneshot(&msg).await? {
649            ClientResp::TableRemoveResp(_) => Ok(()),
650            resp => Err(resp.into()),
651        }
652    }
653
654    /// Replace all rows in this [`Table`] with the input data, coerced to this
655    /// [`Table`]'s existing [`Schema`], notifying any derived [`View`] and
656    /// [`View::on_update`] callbacks.
657    ///
658    /// Calling [`Table::replace`] is an easy way to replace _all_ the data in a
659    /// [`Table`] without losing any derived [`View`] instances or
660    /// [`View::on_update`] callbacks. [`Table::replace`] does _not_ infer
661    /// data types like [`Client::table`] does, rather it _coerces_ input
662    /// data to the `Schema` like [`Table::update`]. If you need a [`Table`]
663    /// with a different `Schema`, you must create a new one.
664    ///
665    /// # Examples
666    ///
667    /// ```no_run
668    /// # use perspective_client::{Table, UpdateData};
669    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
670    /// # let table: Table = todo!();
671    /// let data = UpdateData::Csv("x,y\n1,2".into());
672    /// table.replace(data).await?;
673    /// # Ok(()) }
674    /// ```
675    pub async fn replace(&self, input: UpdateData) -> ClientResult<()> {
676        let msg = self.client_message(ClientReq::TableReplaceReq(TableReplaceReq {
677            data: Some(input.into()),
678        }));
679
680        match self.client.oneshot(&msg).await? {
681            ClientResp::TableReplaceResp(_) => Ok(()),
682            resp => Err(resp.into()),
683        }
684    }
685
686    /// Updates the rows of this table and any derived [`View`] instances.
687    ///
688    /// Calling [`Table::update`] will trigger the [`View::on_update`] callbacks
689    /// register to derived [`View`], and the call itself will not resolve until
690    /// _all_ derived [`View`]'s are notified.
691    ///
692    /// When updating a [`Table`] with an `index`, [`Table::update`] supports
693    /// partial updates, by omitting columns from the update data.
694    ///
695    /// # Arguments
696    ///
697    /// - `input` - The input data for this [`Table`]. The schema of a [`Table`]
698    ///   is immutable after creation, so this method cannot be called with a
699    ///   schema.
700    /// - `options` - Options for this update step - see [`UpdateOptions`].
701    ///
702    /// # Examples
703    ///
704    /// ```no_run
705    /// # use perspective_client::{Table, UpdateData, UpdateOptions};
706    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
707    /// # let table: Table = todo!();
708    /// let data = UpdateData::Csv("x,y\n1,2".into());
709    /// let opts = UpdateOptions::default();
710    /// table.update(data, opts).await?;
711    /// # Ok(()) }
712    /// ```
713    pub async fn update(&self, input: UpdateData, options: UpdateOptions) -> ClientResult<()> {
714        let msg = self.client_message(ClientReq::TableUpdateReq(TableUpdateReq {
715            data: Some(input.into()),
716            port_id: options.port_id.unwrap_or(0),
717        }));
718
719        match self.client.oneshot(&msg).await? {
720            ClientResp::TableUpdateResp(_) => Ok(()),
721            resp => Err(resp.into()),
722        }
723    }
724
725    /// Validate a complete [`ViewConfigUpdate`] against this table and report
726    /// the schema a [`View`] built from it would have, without creating one.
727    pub async fn describe(
728        &self,
729        config: ViewConfigUpdate,
730    ) -> ClientResult<Result<Description, DescribeError>> {
731        let msg = self.client_message(ClientReq::TableDescribeReq(TableDescribeReq {
732            config: Some(config.into()),
733        }));
734
735        let resp = match self.client.oneshot(&msg).await? {
736            ClientResp::TableDescribeResp(resp) => resp,
737            resp => return Err(resp.into()),
738        };
739
740        let expression_schema = decode_schema(resp.expression_schema)?;
741        let has_errors = !resp.expression_errors.is_empty();
742        Ok(match (resp.result, has_errors) {
743            (Some(table_describe_resp::Result::View(view)), false) => Ok(Description {
744                expression_schema,
745                view_schema: decode_schema(view.schema)?,
746            }),
747            (Some(table_describe_resp::Result::ConfigError(msg)), false) => {
748                Err(DescribeError::Config(msg))
749            },
750            (None, true) => Err(DescribeError::Expressions {
751                expression_schema,
752                errors: resp.expression_errors,
753            }),
754            (result, _) => {
755                return Err(ClientError::Unknown(format!(
756                    "Malformed describe response: result={:?} expression_errors={:?}",
757                    result.is_some(),
758                    resp.expression_errors.keys().collect::<Vec<_>>()
759                )));
760            },
761        })
762    }
763
764    /// Validates the given expressions.
765    pub async fn validate_expressions(
766        &self,
767        expressions: Expressions,
768    ) -> ClientResult<ExprValidationResult> {
769        let expression_alias = expressions.0.clone();
770        let config = ViewConfigUpdate {
771            expressions: Some(expressions),
772            columns: Some(vec![]),
773            ..ViewConfigUpdate::default()
774        };
775
776        match self.describe(config).await? {
777            Ok(d) => Ok(ExprValidationResult {
778                expression_schema: d.expression_schema,
779                errors: HashMap::new(),
780                expression_alias,
781            }),
782            Err(DescribeError::Expressions {
783                expression_schema,
784                errors,
785            }) => Ok(ExprValidationResult {
786                expression_schema,
787                errors,
788                expression_alias,
789            }),
790            Err(DescribeError::Config(msg)) => Err(ClientError::Internal(msg)),
791        }
792    }
793
794    /// Create a new [`View`] from this table with a specified
795    /// [`ViewConfigUpdate`].
796    ///
797    /// See [`View`] struct.
798    ///
799    /// # Examples
800    ///
801    /// ```no_run
802    /// # use std::collections::HashMap;
803    /// # use perspective_client::Table;
804    /// # use perspective_client::config::*;
805    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
806    /// # let table: Table = todo!();
807    /// let view = table
808    ///     .view(Some(ViewConfigUpdate {
809    ///         columns: Some(vec![Some("Sales".into())]),
810    ///         aggregates: Some(HashMap::from_iter(vec![("Sales".into(), "sum".into())])),
811    ///         group_by: Some(vec!["Region".into(), "Country".into()]),
812    ///         filter: Some(vec![Filter::new("Category", "in", &[
813    ///             "Furniture",
814    ///             "Technology",
815    ///         ])]),
816    ///         ..ViewConfigUpdate::default()
817    ///     }))
818    ///     .await?;
819    /// # Ok(()) }
820    /// ```
821    pub async fn view(&self, config: Option<ViewConfigUpdate>) -> ClientResult<View> {
822        let view_name = randid();
823        let msg = Request {
824            msg_id: self.client.gen_id(),
825            entity_id: self.name.clone(),
826            client_req: ClientReq::TableMakeViewReq(TableMakeViewReq {
827                view_id: view_name.clone(),
828                config: config.map(|x| x.into()),
829            })
830            .into(),
831        };
832
833        match self.client.oneshot(&msg).await? {
834            ClientResp::TableMakeViewResp(TableMakeViewResp { view_id })
835                if view_id == view_name =>
836            {
837                Ok(View::new_with_source(
838                    view_name,
839                    self.client.clone(),
840                    ViewSource {
841                        options: self.options.clone(),
842                    },
843                ))
844            },
845            resp => Err(resp.into()),
846        }
847    }
848}