perspective_client/virtual_server/
server.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;
14
15use indexmap::{IndexMap, IndexSet};
16use prost::Message as ProstMessage;
17use prost::bytes::{Bytes, BytesMut};
18
19use super::data::RowPathStyle;
20use super::error::VirtualServerError;
21use super::generic_sql_model::{column_path_source, sort_column_paths};
22use super::handler::VirtualServerHandler;
23use crate::config::{ViewConfig, ViewConfigUpdate};
24use crate::proto::response::ClientResp;
25use crate::proto::{
26    ColumnPathLevel, ColumnType, GetFeaturesResp, GetHostedTablesResp, MakeTableResp, Request,
27    Response, Scalar, ServerError, TableDeleteResp, TableDescribeResp, TableMakePortResp,
28    TableMakeViewResp, TableOnDeleteResp, TableRemoveDeleteResp, TableSchemaResp, TableSizeResp,
29    ViewColumnPathsResp, ViewDeleteResp, ViewDescription, ViewDimensionsResp,
30    ViewExpressionSchemaResp, ViewGetConfigResp, ViewGetMinMaxResp, ViewOnDeleteResp,
31    ViewOnRemoveResp, ViewOnUpdateResp, ViewRemoveDeleteResp, ViewRemoveOnRemoveResp,
32    ViewRemoveOnUpdateResp, ViewSchemaResp, ViewToArrowResp, ViewToColumnsStringResp,
33    ViewToCsvResp, ViewToNdjsonStringResp, ViewToRowsStringResp, scalar, table_describe_resp,
34};
35use crate::table::{DescribeError, Description};
36
37macro_rules! respond {
38    ($msg:ident, $name:ident { $($rest:tt)* }) => {{
39        let mut resp = BytesMut::new();
40        let resp2 = ClientResp::$name($name {
41            $($rest)*
42        });
43
44        Response {
45            msg_id: $msg.msg_id,
46            entity_id: $msg.entity_id,
47            client_resp: Some(resp2),
48        }.encode(&mut resp).map_err(VirtualServerError::EncodeError)?;
49
50        resp.freeze()
51    }};
52}
53
54/// A virtual server that processes Perspective protocol messages.
55///
56/// `VirtualServer` acts as a bridge between the Perspective protocol and a
57/// custom data backend. It handles protocol decoding/encoding and delegates
58/// actual data operations to the provided [`VirtualServerHandler`].
59pub struct VirtualServer<T: VirtualServerHandler> {
60    handler: T,
61    view_to_table: IndexMap<String, String>,
62    view_configs: IndexMap<String, ViewConfig>,
63    view_schemas: IndexMap<String, IndexMap<String, ColumnType>>,
64
65    /// The tables made from views, which this server created and may drop.
66    view_tables: IndexSet<String>,
67
68    /// Per-view `table_describe` answers, computed LAZILY on the first
69    /// `ViewExpressionSchemaReq` for that view — never on view creation.
70    view_descriptions: IndexMap<String, Description>,
71}
72
73impl<T: VirtualServerHandler> VirtualServer<T> {
74    /// Creates a new virtual server with the given handler.
75    pub fn new(handler: T) -> Self {
76        Self {
77            handler,
78            view_configs: IndexMap::default(),
79            view_to_table: IndexMap::default(),
80            view_schemas: IndexMap::default(),
81            view_tables: IndexSet::default(),
82            view_descriptions: IndexMap::default(),
83        }
84    }
85
86    /// Processes a Perspective protocol request and returns the response.
87    ///
88    /// Decodes the incoming protobuf message, dispatches to the appropriate
89    /// handler method, and encodes the response.
90    pub async fn handle_request(
91        &mut self,
92        bytes: Bytes,
93    ) -> Result<Bytes, VirtualServerError<T::Error>> {
94        let msg = Request::decode(bytes).map_err(VirtualServerError::DecodeError)?;
95        tracing::debug!(
96            "Handling request: entity_id={}, req={:?}",
97            msg.entity_id,
98            msg.client_req
99        );
100
101        match self.internal_handle_request(msg.clone()).await {
102            Ok(resp) => Ok(resp),
103            Err(err) => {
104                tracing::error!("{}", err);
105                Ok(respond!(msg, ServerError {
106                    message: err.to_string(),
107                    status_code: 0
108                }))
109            },
110        }
111    }
112
113    /// Deletes every view this server still holds, as closing a native
114    /// `Session` does, returning the first handler error after attempting all.
115    pub async fn close(&mut self) -> Result<(), VirtualServerError<T::Error>> {
116        let mut result = Ok(());
117        for view_id in self.view_configs.keys().cloned().collect::<Vec<_>>() {
118            if let Err(err) = self.handler.view_delete(&view_id).await
119                && result.is_ok()
120            {
121                result = Err(err.into());
122            }
123
124            self.view_to_table.shift_remove(&view_id);
125            self.view_configs.shift_remove(&view_id);
126            self.view_schemas.shift_remove(&view_id);
127            self.view_descriptions.shift_remove(&view_id);
128        }
129
130        result
131    }
132
133    async fn get_cached_view_schema(
134        &mut self,
135        entity_id: &str,
136        to_psp_format: bool,
137    ) -> Result<IndexMap<String, ColumnType>, VirtualServerError<T::Error>> {
138        if !self.view_schemas.contains_key(entity_id) {
139            self.view_schemas.insert(
140                entity_id.to_string(),
141                self.handler
142                    .view_schema(entity_id, self.view_configs.get(entity_id).unwrap())
143                    .await?,
144            );
145        }
146
147        if to_psp_format {
148            // `view.schema()` is keyed by *source* column name, matching the
149            // native engine, while the cached schema is keyed by the view's
150            // actual (possibly pivoted-path) SQL column names.
151            let config = self.view_configs.get(entity_id).unwrap();
152            Ok(self
153                .view_schemas
154                .get(entity_id)
155                .unwrap()
156                .iter()
157                .map(|(k, v)| {
158                    let name = column_path_source(k, config)
159                        .map(|(_, col)| col.to_string())
160                        .unwrap_or_else(|| k.clone());
161
162                    (name, *v)
163                })
164                .collect())
165        } else {
166            Ok(self.view_schemas.get(entity_id).cloned().unwrap())
167        }
168    }
169
170    async fn internal_handle_request(
171        &mut self,
172        msg: Request,
173    ) -> Result<Bytes, VirtualServerError<T::Error>> {
174        use crate::proto::request::ClientReq::*;
175        let resp = match msg.client_req.unwrap() {
176            GetFeaturesReq(_) => {
177                let features = self.handler.get_features().await?;
178                respond!(msg, GetFeaturesResp { ..features.into() })
179            },
180            GetHostedTablesReq(_) => {
181                respond!(msg, GetHostedTablesResp {
182                    table_infos: self.handler.get_hosted_tables().await?
183                })
184            },
185            TableSchemaReq(_) => {
186                respond!(msg, TableSchemaResp {
187                    schema: Some(crate::proto::Schema {
188                        schema: self
189                            .handler
190                            .table_schema(msg.entity_id.as_str())
191                            .await?
192                            .iter()
193                            .map(|x| crate::proto::schema::KeyTypePair {
194                                name: x.0.to_string(),
195                                r#type: *x.1 as i32,
196                            })
197                            .collect()
198                    })
199                })
200            },
201            TableMakePortReq(req) => {
202                respond!(msg, TableMakePortResp {
203                    port_id: self.handler.table_make_port(&req).await?
204                })
205            },
206            TableMakeViewReq(req) => {
207                self.view_to_table
208                    .insert(req.view_id.clone(), msg.entity_id.clone());
209
210                let mut config: ViewConfigUpdate = req.config.clone().unwrap_or_default().into();
211                if let Some(group_by) = &config.group_by
212                    && let Some(duplicate) = group_by
213                        .iter()
214                        .enumerate()
215                        .find(|(idx, col)| group_by[..*idx].contains(*col))
216                {
217                    return Err(VirtualServerError::Other(format!(
218                        "Duplicate column `{}` in `group_by`",
219                        duplicate.1
220                    )));
221                }
222
223                // An UNORDERED store has no natural row order to fall back
224                // on, so every window must carry an explicit `order_by`.
225                if let Some(windows) = &config.windows
226                    && windows.values().any(|w| w.order_by.is_none())
227                    && self.handler.get_features().await?.unordered
228                {
229                    return Err(VirtualServerError::Other(
230                        "This data store is unordered - windows require an explicit `order_by`"
231                            .to_string(),
232                    ));
233                }
234
235                let bytes = respond!(msg, TableMakeViewResp {
236                    view_id: self
237                        .handler
238                        .table_make_view(msg.entity_id.as_str(), req.view_id.as_str(), &mut config)
239                        .await?
240                });
241
242                self.view_configs.insert(req.view_id.clone(), config.into());
243                bytes
244            },
245            TableSizeReq(_) => {
246                respond!(msg, TableSizeResp {
247                    size: self.handler.table_size(msg.entity_id.as_str()).await?
248                })
249            },
250            TableDescribeReq(req) => {
251                let config: ViewConfig = req.config.unwrap_or_default().into();
252                let verdict = self
253                    .handler
254                    .table_describe(msg.entity_id.as_str(), &config)
255                    .await?;
256
257                let (expression_schema, expression_errors, result) = match verdict {
258                    Ok(d) => (
259                        d.expression_schema,
260                        HashMap::new(),
261                        Some(table_describe_resp::Result::View(ViewDescription {
262                            schema: d
263                                .view_schema
264                                .into_iter()
265                                .map(|(x, y)| (x, y as i32))
266                                .collect(),
267                        })),
268                    ),
269                    Err(DescribeError::Expressions {
270                        expression_schema,
271                        errors,
272                    }) => (expression_schema, errors, None),
273                    Err(DescribeError::Config(msg)) => (
274                        HashMap::new(),
275                        HashMap::new(),
276                        Some(table_describe_resp::Result::ConfigError(msg)),
277                    ),
278                };
279
280                respond!(msg, TableDescribeResp {
281                    expression_schema: expression_schema
282                        .into_iter()
283                        .map(|(x, y)| (x, y as i32))
284                        .collect(),
285                    expression_errors,
286                    result,
287                })
288            },
289            ViewSchemaReq(_) => {
290                respond!(msg, ViewSchemaResp {
291                    schema: self
292                        .get_cached_view_schema(&msg.entity_id, true)
293                        .await?
294                        .into_iter()
295                        .map(|(x, y)| (x.to_string(), y as i32))
296                        .collect()
297                })
298            },
299            ViewDimensionsReq(_) => {
300                let view_id = &msg.entity_id;
301                let table_id = self
302                    .view_to_table
303                    .get(view_id)
304                    .ok_or_else(|| VirtualServerError::UnknownViewId(view_id.to_string()))?;
305
306                let num_table_rows = self.handler.table_size(table_id).await?;
307                let num_table_columns = self.handler.table_column_size(table_id).await? as u32;
308                let config = self.view_configs.get(view_id).unwrap();
309                let num_view_columns = self.handler.view_column_size(view_id, config).await? as u32;
310                let num_view_rows = self.handler.view_size(view_id).await?;
311                let resp = ViewDimensionsResp {
312                    num_table_columns,
313                    num_table_rows,
314                    num_view_columns,
315                    num_view_rows,
316                };
317
318                respond!(msg, ViewDimensionsResp { ..resp })
319            },
320            ViewGetConfigReq(_) => {
321                respond!(msg, ViewGetConfigResp {
322                    config: Some(
323                        ViewConfigUpdate::from(
324                            self.view_configs.get(&msg.entity_id).unwrap().clone()
325                        )
326                        .into()
327                    )
328                })
329            },
330            ViewExpressionSchemaReq(_) => {
331                let view_id = msg.entity_id.clone();
332                if !self.view_descriptions.contains_key(&view_id) {
333                    let table_id = self
334                        .view_to_table
335                        .get(&view_id)
336                        .cloned()
337                        .ok_or_else(|| VirtualServerError::UnknownViewId(view_id.clone()))?;
338
339                    let config = self.view_configs.get(&view_id).unwrap().clone();
340                    let description = self
341                        .handler
342                        .table_describe(&table_id, &config)
343                        .await?
344                        .map_err(|e| VirtualServerError::Other(e.to_string()))?;
345
346                    self.view_descriptions.insert(view_id.clone(), description);
347                }
348
349                let schema = self.view_descriptions[&view_id]
350                    .expression_schema
351                    .iter()
352                    .map(|(name, ty)| (name.clone(), *ty as i32))
353                    .collect();
354
355                respond!(msg, ViewExpressionSchemaResp { schema })
356            },
357            ViewColumnPathsReq(view_column_paths_req) => {
358                let config = self.view_configs.get(&msg.entity_id).unwrap();
359                let mut paths: Vec<String> = self
360                    .handler
361                    .view_schema(msg.entity_id.as_str(), config)
362                    .await?
363                    .keys()
364                    .cloned()
365                    .collect();
366
367                if !config.split_by.is_empty() {
368                    sort_column_paths(&mut paths, config);
369                }
370
371                let start = view_column_paths_req.start_col.unwrap_or(0) as usize;
372                let end = view_column_paths_req
373                    .end_col
374                    .map_or(paths.len(), |x| x as usize);
375
376                let paths = paths
377                    .into_iter()
378                    .skip(start)
379                    .take(end.saturating_sub(start))
380                    .collect::<Vec<_>>();
381
382                let depth = config.split_by.len() + 1;
383                let mut area = vec![
384                    ColumnPathLevel {
385                        values: Vec::with_capacity(paths.len())
386                    };
387                    depth
388                ];
389
390                for path in paths {
391                    let mut split = path.splitn(depth, '|');
392                    let mut levels = Vec::with_capacity(depth);
393                    for _ in 0..depth {
394                        match split.next() {
395                            Some(x) => levels.push(Some(x.to_owned())),
396                            None => levels.push(None),
397                        }
398                    }
399
400                    let name = levels
401                        .iter()
402                        .rposition(|x| x.is_some())
403                        .map(|i| levels[i].take().unwrap())
404                        .unwrap_or_default();
405
406                    for (level, value) in area.iter_mut().zip(levels).take(depth - 1) {
407                        level.values.push(Scalar {
408                            scalar: value.map(scalar::Scalar::String),
409                        });
410                    }
411
412                    area[depth - 1].values.push(Scalar {
413                        scalar: Some(scalar::Scalar::String(name)),
414                    });
415                }
416
417                respond!(msg, ViewColumnPathsResp { area })
418            },
419            ViewToArrowReq(view_to_arrow_req) => {
420                let viewport = view_to_arrow_req.viewport.unwrap();
421                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
422                let config = self.view_configs.get(&msg.entity_id).unwrap();
423                let mut cols = self
424                    .handler
425                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
426                    .await?;
427
428                let arrow = cols
429                    .render_to_arrow_ipc()
430                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
431
432                respond!(msg, ViewToArrowResp { arrow })
433            },
434            ViewToCsvReq(view_to_csv_req) => {
435                let viewport = view_to_csv_req.viewport.unwrap();
436                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
437                let config = self.view_configs.get(&msg.entity_id).unwrap();
438                let mut cols = self
439                    .handler
440                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
441                    .await?;
442
443                let rows = cols.render_to_rows(RowPathStyle::PerLevel);
444                let mut csv = String::new();
445                if let Some(first_row) = rows.first() {
446                    let headers: Vec<&str> = first_row.keys().map(|k| k.as_str()).collect();
447                    csv.push_str(&headers.join(","));
448                    csv.push('\n');
449                }
450
451                for row in &rows {
452                    let values: Vec<String> = row
453                        .values()
454                        .map(|cell| serde_json::to_string(cell).unwrap_or_default())
455                        .collect();
456                    csv.push_str(&values.join(","));
457                    csv.push('\n');
458                }
459
460                respond!(msg, ViewToCsvResp { csv })
461            },
462            ViewToNdjsonStringReq(view_to_ndjson_req) => {
463                let viewport = view_to_ndjson_req.viewport.unwrap();
464                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
465                let config = self.view_configs.get(&msg.entity_id).unwrap();
466                let mut cols = self
467                    .handler
468                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
469                    .await?;
470
471                let rows = cols.render_to_rows(RowPathStyle::Sidecar);
472                let ndjson_string = rows
473                    .iter()
474                    .map(serde_json::to_string)
475                    .collect::<Result<Vec<_>, _>>()
476                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?
477                    .join("\n");
478
479                respond!(msg, ViewToNdjsonStringResp { ndjson_string })
480            },
481            ViewToRowsStringReq(view_to_rows_string_req) => {
482                let viewport = view_to_rows_string_req.viewport.unwrap();
483                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
484                let config = self.view_configs.get(&msg.entity_id).unwrap();
485                let mut cols = self
486                    .handler
487                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
488                    .await?;
489
490                let rows = cols.render_to_rows(RowPathStyle::Sidecar);
491                let json_string = serde_json::to_string(&rows)
492                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?;
493
494                respond!(msg, ViewToRowsStringResp { json_string })
495            },
496            ViewToColumnsStringReq(view_to_columns_string_req) => {
497                let viewport = view_to_columns_string_req.viewport.unwrap();
498                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
499                let config = self.view_configs.get(&msg.entity_id).unwrap();
500                let mut cols = self
501                    .handler
502                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
503                    .await?;
504
505                let json_string = cols
506                    .render_to_columns_json(
507                        RowPathStyle::Sidecar,
508                        view_to_columns_string_req.id.unwrap_or_default(),
509                    )
510                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
511
512                respond!(msg, ViewToColumnsStringResp { json_string })
513            },
514            ViewDeleteReq(_) => {
515                self.handler.view_delete(msg.entity_id.as_str()).await?;
516                self.view_to_table.shift_remove(&msg.entity_id);
517                self.view_configs.shift_remove(&msg.entity_id);
518                self.view_descriptions.shift_remove(&msg.entity_id);
519                respond!(msg, ViewDeleteResp {})
520            },
521            MakeTableReq(req) => {
522                let data = req.data.as_ref().unwrap();
523                if let Some(crate::proto::make_table_data::Data::FromView(view_id)) = &data.data {
524                    if !self.handler.get_features().await?.view_derivations {
525                        return Err(VirtualServerError::Other(
526                            "This data source cannot derive a Table from a View".to_string(),
527                        ));
528                    }
529
530                    let options = req.options.clone().unwrap_or_default();
531                    if options.make_table_type.is_some() {
532                        return Err(VirtualServerError::Other(
533                            "`index` and `limit` cannot be set on a Table made from a View"
534                                .to_string(),
535                        ));
536                    }
537
538                    let config = self.view_configs.get(view_id).ok_or_else(|| {
539                        VirtualServerError::Other(format!("View \"{}\" does not exist", view_id))
540                    })?;
541
542                    let schema = options
543                        .view_schema
544                        .map(|schema| {
545                            schema
546                                .schema
547                                .into_iter()
548                                .map(|x| Ok((x.name, ColumnType::try_from(x.r#type)?)))
549                                .collect::<Result<IndexMap<_, _>, prost::DecodeError>>()
550                        })
551                        .transpose()
552                        .map_err(|e| VirtualServerError::Other(e.to_string()))?;
553
554                    self.handler
555                        .view_make_table(view_id, &msg.entity_id, config, schema.as_ref())
556                        .await?;
557
558                    self.view_tables.insert(msg.entity_id.clone());
559                } else {
560                    self.handler.make_table(&msg.entity_id, data).await?;
561                }
562
563                respond!(msg, MakeTableResp {})
564            },
565            ViewGetMinMaxReq(req) => {
566                let config = self.view_configs.get(&msg.entity_id).unwrap();
567                let (min, max) = self
568                    .handler
569                    .view_get_min_max(&msg.entity_id, &req.column_name, config)
570                    .await?;
571                respond!(msg, ViewGetMinMaxResp {
572                    min: Some(min.into()),
573                    max: Some(max.into()),
574                })
575            },
576
577            TableDeleteReq(_) if self.view_tables.contains(&msg.entity_id) => {
578                self.handler.view_delete(msg.entity_id.as_str()).await?;
579                self.view_tables.shift_remove(&msg.entity_id);
580                respond!(msg, TableDeleteResp {})
581            },
582
583            // Stub implementations for callback/update requests that VirtualServer doesn't support
584            TableOnDeleteReq(_) => {
585                respond!(msg, TableOnDeleteResp {})
586            },
587            ViewOnRemoveReq(_) => {
588                respond!(msg, ViewOnRemoveResp {
589                    indices: None,
590                    port_id: 0
591                })
592            },
593            ViewRemoveOnRemoveReq(_) => {
594                respond!(msg, ViewRemoveOnRemoveResp {})
595            },
596            ViewOnUpdateReq(_) => {
597                respond!(msg, ViewOnUpdateResp {
598                    delta: None,
599                    port_id: 0
600                })
601            },
602            ViewOnDeleteReq(_) => {
603                respond!(msg, ViewOnDeleteResp {})
604            },
605            ViewRemoveOnUpdateReq(_) => {
606                respond!(msg, ViewRemoveOnUpdateResp {})
607            },
608            TableRemoveDeleteReq(_) => {
609                respond!(msg, TableRemoveDeleteResp {})
610            },
611            ViewRemoveDeleteReq(_) => {
612                respond!(msg, ViewRemoveDeleteResp {})
613            },
614            x => {
615                // Return an error response instead of empty bytes
616                return Err(VirtualServerError::Other(format!(
617                    "Unhandled request: {:?}",
618                    x
619                )));
620            },
621        };
622
623        Ok(resp)
624    }
625}
perspective_client/virtual_server/server.rs

perspective_client/virtual_server/
server.rs
