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(_) => {
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                respond!(msg, ViewColumnPathsResp { paths })
322            },
323            ViewToArrowReq(view_to_arrow_req) => {
324                let viewport = view_to_arrow_req.viewport.unwrap();
325                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
326                let config = self.view_configs.get(&msg.entity_id).unwrap();
327                let mut cols = self
328                    .handler
329                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
330                    .await?;
331
332                let arrow = cols
333                    .render_to_arrow_ipc()
334                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
335
336                respond!(msg, ViewToArrowResp { arrow })
337            },
338            ViewToCsvReq(view_to_csv_req) => {
339                let viewport = view_to_csv_req.viewport.unwrap();
340                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
341                let config = self.view_configs.get(&msg.entity_id).unwrap();
342                let mut cols = self
343                    .handler
344                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
345                    .await?;
346
347                let rows = cols.render_to_rows(RowPathStyle::PerLevel);
348                let mut csv = String::new();
349                if let Some(first_row) = rows.first() {
350                    let headers: Vec<&str> = first_row.keys().map(|k| k.as_str()).collect();
351                    csv.push_str(&headers.join(","));
352                    csv.push('\n');
353                }
354
355                for row in &rows {
356                    let values: Vec<String> = row
357                        .values()
358                        .map(|cell| serde_json::to_string(cell).unwrap_or_default())
359                        .collect();
360                    csv.push_str(&values.join(","));
361                    csv.push('\n');
362                }
363
364                respond!(msg, ViewToCsvResp { csv })
365            },
366            ViewToNdjsonStringReq(view_to_ndjson_req) => {
367                let viewport = view_to_ndjson_req.viewport.unwrap();
368                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
369                let config = self.view_configs.get(&msg.entity_id).unwrap();
370                let mut cols = self
371                    .handler
372                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
373                    .await?;
374
375                let rows = cols.render_to_rows(RowPathStyle::PerLevel);
376                let ndjson_string = rows
377                    .iter()
378                    .map(serde_json::to_string)
379                    .collect::<Result<Vec<_>, _>>()
380                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?
381                    .join("\n");
382
383                respond!(msg, ViewToNdjsonStringResp { ndjson_string })
384            },
385            ViewToRowsStringReq(view_to_rows_string_req) => {
386                let viewport = view_to_rows_string_req.viewport.unwrap();
387                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
388                let config = self.view_configs.get(&msg.entity_id).unwrap();
389                let mut cols = self
390                    .handler
391                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
392                    .await?;
393
394                let rows = cols.render_to_rows(RowPathStyle::Sidecar);
395                let json_string = serde_json::to_string(&rows)
396                    .map_err(|e| VirtualServerError::InvalidJSON(std::sync::Arc::new(e)))?;
397
398                respond!(msg, ViewToRowsStringResp { json_string })
399            },
400            ViewToColumnsStringReq(view_to_columns_string_req) => {
401                let viewport = view_to_columns_string_req.viewport.unwrap();
402                let schema = self.get_cached_view_schema(&msg.entity_id, false).await?;
403                let config = self.view_configs.get(&msg.entity_id).unwrap();
404                let mut cols = self
405                    .handler
406                    .view_get_data(msg.entity_id.as_str(), config, &schema, &viewport)
407                    .await?;
408
409                let json_string = cols
410                    .render_to_columns_json(
411                        RowPathStyle::Sidecar,
412                        view_to_columns_string_req.id.unwrap_or_default(),
413                    )
414                    .map_err(|e| VirtualServerError::Other(e.to_string()))?;
415
416                respond!(msg, ViewToColumnsStringResp { json_string })
417            },
418            ViewDeleteReq(_) => {
419                self.handler.view_delete(msg.entity_id.as_str()).await?;
420                self.view_to_table.shift_remove(&msg.entity_id);
421                self.view_configs.shift_remove(&msg.entity_id);
422                respond!(msg, ViewDeleteResp {})
423            },
424            MakeTableReq(req) => {
425                self.handler
426                    .make_table(&msg.entity_id, req.data.as_ref().unwrap())
427                    .await?;
428                respond!(msg, MakeTableResp {})
429            },
430            ViewGetMinMaxReq(req) => {
431                let config = self.view_configs.get(&msg.entity_id).unwrap();
432                let (min, max) = self
433                    .handler
434                    .view_get_min_max(&msg.entity_id, &req.column_name, config)
435                    .await?;
436                respond!(msg, ViewGetMinMaxResp {
437                    min: Some(min.into()),
438                    max: Some(max.into()),
439                })
440            },
441
442            // Stub implementations for callback/update requests that VirtualServer doesn't support
443            TableOnDeleteReq(_) => {
444                respond!(msg, TableOnDeleteResp {})
445            },
446            ViewOnUpdateReq(_) => {
447                respond!(msg, ViewOnUpdateResp {
448                    delta: None,
449                    port_id: 0
450                })
451            },
452            ViewOnDeleteReq(_) => {
453                respond!(msg, ViewOnDeleteResp {})
454            },
455            ViewRemoveOnUpdateReq(_) => {
456                respond!(msg, ViewRemoveOnUpdateResp {})
457            },
458            TableRemoveDeleteReq(_) => {
459                respond!(msg, TableRemoveDeleteResp {})
460            },
461            ViewRemoveDeleteReq(_) => {
462                respond!(msg, ViewRemoveDeleteResp {})
463            },
464            x => {
465                // Return an error response instead of empty bytes
466                return Err(VirtualServerError::Other(format!(
467                    "Unhandled request: {:?}",
468                    x
469                )));
470            },
471        };
472
473        Ok(resp)
474    }
475}
perspective_client/virtual_server/server.rs

perspective_client/virtual_server/
server.rs
