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;
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::table_validate_expr_resp::ExprValidationError;
26use crate::proto::{
27    ColumnType, GetFeaturesResp, GetHostedTablesResp, MakeTableResp, Request, Response,
28    ServerError, TableMakePortResp, TableMakeViewResp, TableOnDeleteResp, TableRemoveDeleteResp,
29    TableSchemaResp, TableSizeResp, TableValidateExprResp, ViewColumnPathsResp, ViewDeleteResp,
30    ViewDimensionsResp, ViewExpressionSchemaResp, ViewGetConfigResp, ViewGetMinMaxResp,
31    ViewOnDeleteResp, ViewOnUpdateResp, ViewRemoveDeleteResp, ViewRemoveOnUpdateResp,
32    ViewSchemaResp, ViewToArrowResp, ViewToColumnsStringResp, ViewToCsvResp,
33    ViewToNdjsonStringResp, ViewToRowsStringResp,
34};
35
36macro_rules! respond {
37    ($msg:ident, $name:ident { $($rest:tt)* }) => {{
38        let mut resp = BytesMut::new();
39        let resp2 = ClientResp::$name($name {
40            $($rest)*
41        });
42
43        Response {
44            msg_id: $msg.msg_id,
45            entity_id: $msg.entity_id,
46            client_resp: Some(resp2),
47        }.encode(&mut resp).map_err(VirtualServerError::EncodeError)?;
48
49        resp.freeze()
50    }};
51}
52
53/// A virtual server that processes Perspective protocol messages.
54///
55/// `VirtualServer` acts as a bridge between the Perspective protocol and a
56/// custom data backend. It handles protocol decoding/encoding and delegates
57/// actual data operations to the provided [`VirtualServerHandler`].
58pub struct VirtualServer<T: VirtualServerHandler> {
59    handler: T,
60    view_to_table: IndexMap<String, String>,
61    view_configs: IndexMap<String, ViewConfig>,
62    view_schemas: IndexMap<String, IndexMap<String, ColumnType>>,
63}
64
65impl<T: VirtualServerHandler> VirtualServer<T> {
66    /// Creates a new virtual server with the given handler.
67    pub fn new(handler: T) -> Self {
68        Self {
69            handler,
70            view_configs: IndexMap::default(),
71            view_to_table: IndexMap::default(),
72            view_schemas: IndexMap::default(),
73        }
74    }
75
76    /// Processes a Perspective protocol request and returns the response.
77    ///
78    /// Decodes the incoming protobuf message, dispatches to the appropriate
79    /// handler method, and encodes the response.
80    pub async fn handle_request(
81        &mut self,
82        bytes: Bytes,
83    ) -> Result<Bytes, VirtualServerError<T::Error>> {
84        let msg = Request::decode(bytes).map_err(VirtualServerError::DecodeError)?;
85        tracing::debug!(
86            "Handling request: entity_id={}, req={:?}",
87            msg.entity_id,
88            msg.client_req
89        );
90
91        match self.internal_handle_request(msg.clone()).await {
92            Ok(resp) => Ok(resp),
93            Err(err) => {
94                tracing::error!("{}", err);
95                Ok(respond!(msg, ServerError {
96                    message: err.to_string(),
97                    status_code: 0
98                }))
99            },
100        }
101    }
102
103    async fn get_cached_view_schema(
104        &mut self,
105        entity_id: &str,
106        to_psp_format: bool,
107    ) -> Result<IndexMap<String, ColumnType>, VirtualServerError<T::Error>> {
108        if !self.view_schemas.contains_key(entity_id) {
109            self.view_schemas.insert(
110                entity_id.to_string(),
111                self.handler
112                    .view_schema(entity_id, self.view_configs.get(entity_id).unwrap())
113                    .await?,
114            );
115        }
116
117        if to_psp_format {
118            // `view.schema()` is keyed by *source* column name, matching the
119            // native engine, while the cached schema is keyed by the view's
120            // actual (possibly pivoted-path) SQL column names.
121            let config = self.view_configs.get(entity_id).unwrap();
122            Ok(self
123                .view_schemas
124                .get(entity_id)
125                .unwrap()
126                .iter()
127                .map(|(k, v)| {
128                    let name = column_path_source(k, config)
129                        .map(|(_, col)| col.to_string())
130                        .unwrap_or_else(|| k.clone());
131
132                    (name, *v)
133                })
134                .collect())
135        } else {
136            Ok(self.view_schemas.get(entity_id).cloned().unwrap())
137        }
138    }
139
140    async fn internal_handle_request(
141        &mut self,
142        msg: Request,
143    ) -> Result<Bytes, VirtualServerError<T::Error>> {
144        use crate::proto::request::ClientReq::*;
145        let resp = match msg.client_req.unwrap() {
146            GetFeaturesReq(_) => {
147                let features = self.handler.get_features().await?;
148                respond!(msg, GetFeaturesResp { ..features.into() })
149            },
150            GetHostedTablesReq(_) => {
151                respond!(msg, GetHostedTablesResp {
152                    table_infos: self.handler.get_hosted_tables().await?
153                })
154            },
155            TableSchemaReq(_) => {
156                respond!(msg, TableSchemaResp {
157                    schema: Some(crate::proto::Schema {
158                        schema: self
159                            .handler
160                            .table_schema(msg.entity_id.as_str())
161                            .await?
162                            .iter()
163                            .map(|x| crate::proto::schema::KeyTypePair {
164                                name: x.0.to_string(),
165                                r#type: *x.1 as i32,
166                            })
167                            .collect()
168                    })
169                })
170            },
171            TableMakePortReq(req) => {
172                respond!(msg, TableMakePortResp {
173                    port_id: self.handler.table_make_port(&req).await?
174                })
175            },
176            TableMakeViewReq(req) => {
177                self.view_to_table
178                    .insert(req.view_id.clone(), msg.entity_id.clone());
179
180                let mut config: ViewConfigUpdate = req.config.clone().unwrap_or_default().into();
181
182                // An UNORDERED store has no natural row order to fall back
183                // on, so every window must carry an explicit `order_by`.
184                if let Some(windows) = &config.windows
185                    && windows.values().any(|w| w.order_by.is_none())
186                    && self.handler.get_features().await?.unordered
187                {
188                    return Err(VirtualServerError::Other(
189                        "This data store is unordered - windows require an explicit `order_by`"
190                            .to_string(),
191                    ));
192                }
193
194                let bytes = respond!(msg, TableMakeViewResp {
195                    view_id: self
196                        .handler
197                        .table_make_view(msg.entity_id.as_str(), req.view_id.as_str(), &mut config)
198                        .await?
199                });
200
201                self.view_configs.insert(req.view_id.clone(), config.into());
202                bytes
203            },
204            TableSizeReq(_) => {
205                respond!(msg, TableSizeResp {
206                    size: self.handler.table_size(msg.entity_id.as_str()).await?
207                })
208            },
209            TableValidateExprReq(req) => {
210                let mut expression_schema = HashMap::<String, i32>::default();
211                let mut expression_alias = HashMap::<String, String>::default();
212                let mut errors = HashMap::<String, ExprValidationError>::default();
213                for (name, ex) in req.column_to_expr.iter() {
214                    let _ = expression_alias.insert(name.clone(), ex.clone());
215                    match self
216                        .handler
217                        .table_validate_expression(&msg.entity_id, ex.as_str())
218                        .await
219                    {
220                        Ok(dtype) => {
221                            let _ = expression_schema.insert(name.clone(), dtype as i32);
222                        },
223                        Err(e) => {
224                            let _ = errors.insert(name.clone(), ExprValidationError {
225                                error_message: format!("{}", e),
226                                line: 0,
227                                column: 0,
228                            });
229                        },
230                    }
231                }
232
233                respond!(msg, TableValidateExprResp {
234                    expression_schema,
235                    errors,
236                    expression_alias,
237                })
238            },
239            ViewSchemaReq(_) => {
240                respond!(msg, ViewSchemaResp {
241                    schema: self
242                        .get_cached_view_schema(&msg.entity_id, true)
243                        .await?
244                        .into_iter()
245                        .map(|(x, y)| (x.to_string(), y as i32))
246                        .collect()
247                })
248            },
249            ViewDimensionsReq(_) => {
250                let view_id = &msg.entity_id;
251                let table_id = self
252                    .view_to_table
253                    .get(view_id)
254                    .ok_or_else(|| VirtualServerError::UnknownViewId(view_id.to_string()))?;
255
256                let num_table_rows = self.handler.table_size(table_id).await?;
257                let num_table_columns = self.handler.table_column_size(table_id).await? as u32;
258                let config = self.view_configs.get(view_id).unwrap();
259                let num_view_columns = self.handler.view_column_size(view_id, config).await? as u32;
260                let num_view_rows = self.handler.view_size(view_id).await?;
261                let resp = ViewDimensionsResp {
262                    num_table_columns,
263                    num_table_rows,
264                    num_view_columns,
265                    num_view_rows,
266                };
267
268                respond!(msg, ViewDimensionsResp { ..resp })
269            },
270            ViewGetConfigReq(_) => {
271                respond!(msg, ViewGetConfigResp {
272                    config: Some(
273                        ViewConfigUpdate::from(
274                            self.view_configs.get(&msg.entity_id).unwrap().clone()
275                        )
276                        .into()
277                    )
278                })
279            },
280            ViewExpressionSchemaReq(_) => {
281                let mut schema = HashMap::<String, i32>::default();
282                let table_id = self.view_to_table.get(&msg.entity_id);
283                for (name, ex) in self
284                    .view_configs
285                    .get(&msg.entity_id)
286                    .unwrap()
287                    .expressions
288                    .iter()
289                {
290                    match self
291                        .handler
292                        .table_validate_expression(table_id.unwrap(), ex.as_str())
293                        .await
294                    {
295                        Ok(dtype) => {
296                            let _ = schema.insert(name.clone(), dtype as i32);
297                        },
298                        Err(_e) => {
299                            // TODO: handle error
300                        },
301                    }
302                }
303
304                let resp = ViewExpressionSchemaResp { schema };
305                respond!(msg, ViewExpressionSchemaResp { ..resp })
306            },
307            ViewColumnPathsReq(view_column_paths_req) => {
308                let config = self.view_configs.get(&msg.entity_id).unwrap();
309                let mut paths: Vec<String> = self
310                    .handler
311                    .view_schema(msg.entity_id.as_str(), config)
312                    .await?
313                    .keys()
314                    .cloned()
315                    .collect();
316
317                if !config.split_by.is_empty() {
318                    sort_column_paths(&mut paths, config);
319                }
320
321                let start = view_column_paths_req.start_col.unwrap_or(0) as usize;
322                let end = view_column_paths_req
323                    .end_col
324                    .map_or(paths.len(), |x| x as usize);
325
326                let paths = paths.into_iter().take(end).skip(start).collect::<Vec<_>>();
327
328                respond!(msg, ViewColumnPathsResp { paths })
329            },
330            ViewToArrowReq(view_to_arrow_req) => {
331                let viewport = view_to_arrow_req.viewport.unwrap();
332                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
333                let config = self.view_configs.get(&msg.entity_id).unwrap();
334                let mut cols = self
335                    .handler
336                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
337                    .await?;
338
339                let arrow = cols
340                    .render_to_arrow_ipc()
341                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
342
343                respond!(msg, ViewToArrowResp { arrow })
344            },
345            ViewToCsvReq(view_to_csv_req) => {
346                let viewport = view_to_csv_req.viewport.unwrap();
347                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
348                let config = self.view_configs.get(&msg.entity_id).unwrap();
349                let mut cols = self
350                    .handler
351                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
352                    .await?;
353
354                let rows = cols.render_to_rows(RowPathStyle::PerLevel);
355                let mut csv = String::new();
356                if let Some(first_row) = rows.first() {
357                    let headers: Vec<&str> = first_row.keys().map(|k| k.as_str()).collect();
358                    csv.push_str(&headers.join(","));
359                    csv.push('\n');
360                }
361
362                for row in &rows {
363                    let values: Vec<String> = row
364                        .values()
365                        .map(|cell| serde_json::to_string(cell).unwrap_or_default())
366                        .collect();
367                    csv.push_str(&values.join(","));
368                    csv.push('\n');
369                }
370
371                respond!(msg, ViewToCsvResp { csv })
372            },
373            ViewToNdjsonStringReq(view_to_ndjson_req) => {
374                let viewport = view_to_ndjson_req.viewport.unwrap();
375                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
376                let config = self.view_configs.get(&msg.entity_id).unwrap();
377                let mut cols = self
378                    .handler
379                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
380                    .await?;
381
382                let rows = cols.render_to_rows(RowPathStyle::PerLevel);
383                let ndjson_string = rows
384                    .iter()
385                    .map(serde_json::to_string)
386                    .collect::<Result<Vec<_>, _>>()
387                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?
388                    .join("\n");
389
390                respond!(msg, ViewToNdjsonStringResp { ndjson_string })
391            },
392            ViewToRowsStringReq(view_to_rows_string_req) => {
393                let viewport = view_to_rows_string_req.viewport.unwrap();
394                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
395                let config = self.view_configs.get(&msg.entity_id).unwrap();
396                let mut cols = self
397                    .handler
398                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
399                    .await?;
400
401                let rows = cols.render_to_rows(RowPathStyle::Sidecar);
402                let json_string = serde_json::to_string(&rows)
403                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?;
404
405                respond!(msg, ViewToRowsStringResp { json_string })
406            },
407            ViewToColumnsStringReq(view_to_columns_string_req) => {
408                let viewport = view_to_columns_string_req.viewport.unwrap();
409                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
410                let config = self.view_configs.get(&msg.entity_id).unwrap();
411                let mut cols = self
412                    .handler
413                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
414                    .await?;
415
416                let json_string = cols
417                    .render_to_columns_json(
418                        RowPathStyle::Sidecar,
419                        view_to_columns_string_req.id.unwrap_or_default(),
420                    )
421                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
422
423                respond!(msg, ViewToColumnsStringResp { json_string })
424            },
425            ViewDeleteReq(_) => {
426                self.handler.view_delete(msg.entity_id.as_str()).await?;
427                self.view_to_table.shift_remove(&msg.entity_id);
428                self.view_configs.shift_remove(&msg.entity_id);
429                respond!(msg, ViewDeleteResp {})
430            },
431            MakeTableReq(req) => {
432                self.handler
433                    .make_table(&msg.entity_id, req.data.as_ref().unwrap())
434                    .await?;
435                respond!(msg, MakeTableResp {})
436            },
437            ViewGetMinMaxReq(req) => {
438                let config = self.view_configs.get(&msg.entity_id).unwrap();
439                let (min, max) = self
440                    .handler
441                    .view_get_min_max(&msg.entity_id, &req.column_name, config)
442                    .await?;
443                respond!(msg, ViewGetMinMaxResp {
444                    min: Some(min.into()),
445                    max: Some(max.into()),
446                })
447            },
448
449            // Stub implementations for callback/update requests that VirtualServer doesn't support
450            TableOnDeleteReq(_) => {
451                respond!(msg, TableOnDeleteResp {})
452            },
453            ViewOnUpdateReq(_) => {
454                respond!(msg, ViewOnUpdateResp {
455                    delta: None,
456                    port_id: 0
457                })
458            },
459            ViewOnDeleteReq(_) => {
460                respond!(msg, ViewOnDeleteResp {})
461            },
462            ViewRemoveOnUpdateReq(_) => {
463                respond!(msg, ViewRemoveOnUpdateResp {})
464            },
465            TableRemoveDeleteReq(_) => {
466                respond!(msg, TableRemoveDeleteResp {})
467            },
468            ViewRemoveDeleteReq(_) => {
469                respond!(msg, ViewRemoveDeleteResp {})
470            },
471            x => {
472                // Return an error response instead of empty bytes
473                return Err(VirtualServerError::Other(format!(
474                    "Unhandled request: {:?}",
475                    x
476                )));
477            },
478        };
479
480        Ok(resp)
481    }
482}
perspective_client/virtual_server/server.rs

perspective_client/virtual_server/
server.rs
