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

perspective_client/virtual_server/
server.rs
