use std::{
collections::HashMap,
sync::{Arc, RwLock},
};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
#[derive(Debug, Default, Clone)]
pub struct SchemaRegistry {
inner: Arc<RwLock<HashMap<String, SchemaRef>>>,
names: Arc<RwLock<HashMap<String, String>>>,
}
impl SchemaRegistry {
pub fn new() -> Self {
Self::default()
}
pub fn register(&self, key: impl Into<String>, schema: SchemaRef) -> &Self {
self.inner.write().unwrap().insert(key.into(), schema);
self
}
pub fn register_vertex_query(&self, query: &str, extra_cols: &[&str]) -> &Self {
let mut fields = vec![Field::new("_gar_id", DataType::Int64, false)];
fields.extend(
extra_cols
.iter()
.map(|&n| Field::new(n, DataType::Utf8, true)),
);
self.register(query, Arc::new(Schema::new(fields)))
}
pub fn register_edge_query(&self, query: &str, extra_cols: &[&str]) -> &Self {
let mut fields = vec![
Field::new("src", DataType::Int64, false),
Field::new("dst", DataType::Int64, false),
];
fields.extend(
extra_cols
.iter()
.map(|&n| Field::new(n, DataType::Utf8, true)),
);
self.register(query, Arc::new(Schema::new(fields)))
}
pub fn register_named(
&self,
name: impl Into<String>,
query: impl Into<String>,
schema: SchemaRef,
) -> &Self {
let query = query.into();
self.names
.write()
.unwrap()
.insert(name.into(), query.clone());
self.register(query, schema)
}
pub fn get(&self, key: &str) -> Option<SchemaRef> {
self.inner.read().unwrap().get(key).cloned()
}
pub fn keys(&self) -> Vec<String> {
self.inner.read().unwrap().keys().cloned().collect()
}
pub fn name_of(&self, query: &str) -> Option<String> {
self.names
.read()
.unwrap()
.iter()
.find(|(_, q)| q.as_str() == query)
.map(|(name, _)| name.clone())
}
pub fn resolve_table(&self, ident: &str) -> Option<String> {
if let Some(query) = self.names.read().unwrap().get(ident) {
return Some(query.clone());
}
if self.inner.read().unwrap().contains_key(ident) {
return Some(ident.to_string());
}
None
}
pub fn tables(&self) -> Vec<(String, SchemaRef)> {
let names = self.names.read().unwrap();
let query_to_name: HashMap<&str, &str> = names
.iter()
.map(|(n, q)| (q.as_str(), n.as_str()))
.collect();
self.inner
.read()
.unwrap()
.iter()
.map(|(query, schema)| {
let table = query_to_name
.get(query.as_str())
.map(|n| n.to_string())
.unwrap_or_else(|| query.clone());
(table, schema.clone())
})
.collect()
}
}
pub fn parse_select_from(sql: &str) -> Option<String> {
let lower = sql.to_ascii_lowercase();
let from = lower.find(" from ")?;
let after = sql[from + 6..].trim();
let token = after
.split(|c: char| c.is_whitespace() || c == ';')
.next()?
.trim();
if token.is_empty() {
return None;
}
let unquoted = token.trim_matches('"').trim_matches('`').trim_matches('\'');
Some(unquoted.to_string())
}