diff --git a/crates/schema-macro/src/lib.rs b/crates/schema-macro/src/lib.rs index 9f5823d..aa7a376 100644 --- a/crates/schema-macro/src/lib.rs +++ b/crates/schema-macro/src/lib.rs @@ -63,7 +63,7 @@ pub fn derive_has_schema(input: TokenStream) -> TokenStream { }; // TODO: Make this a bit more generic, if we were to have more keys here // it would be ugly to extend this return argument. - let (is_key, is_cursor, description) = schema_attrs(&field.attrs); + let (is_key, is_cursor, description, reference) = schema_attrs(&field.attrs); let base = quote! { ::schema::Field::new( #field_name, @@ -73,18 +73,21 @@ pub fn derive_has_schema(input: TokenStream) -> TokenStream { }; // No attributes → the bare constructor; otherwise apply them to a local: // `annotate` mutates in place (returns `()`), `with_description` is owned. - if !is_key && !is_cursor && description.is_none() { + if !is_key && !is_cursor && description.is_none() && reference.is_none() { base } else { let key_call = is_key.then(|| quote! { field.annotate(::schema::Field::KEY, "true"); }); let cursor_call = is_cursor.then(|| quote! { field.annotate(::schema::Field::CURSOR, "true"); }); + let ref_call = + reference.map(|target| quote! { field.annotate(::schema::Field::REF, #target); }); let desc_call = description.map(|text| quote! { field = field.with_description(#text); }); quote! { { let mut field = #base; #key_call #cursor_call + #ref_call #desc_call field } @@ -113,10 +116,11 @@ pub fn derive_has_schema(input: TokenStream) -> TokenStream { /// Read a field's `#[schema(...)]` helper attribute: `(is_key, is_cursor, /// description)` from `key`, `cursor`, and `description = "..."` (any combination /// may be present). -fn schema_attrs(attrs: &[syn::Attribute]) -> (bool, bool, Option) { +fn schema_attrs(attrs: &[syn::Attribute]) -> (bool, bool, Option, Option) { let mut is_key = false; let mut is_cursor = false; let mut description = None; + let mut reference = None; for attr in attrs { if !attr.path().is_ident("schema") { continue; @@ -128,11 +132,13 @@ fn schema_attrs(attrs: &[syn::Attribute]) -> (bool, bool, Option) { is_cursor = true; } else if meta.path.is_ident("description") { description = Some(meta.value()?.parse::()?.value()); + } else if meta.path.is_ident("references") { + reference = Some(meta.value()?.parse::()?.value()); } Ok(()) }); } - (is_key, is_cursor, description) + (is_key, is_cursor, description, reference) } /// If `ty` is `Option`, return `Inner`. diff --git a/crates/schema/src/lib.rs b/crates/schema/src/lib.rs index 28a06b2..bd47c6b 100644 --- a/crates/schema/src/lib.rs +++ b/crates/schema/src/lib.rs @@ -42,6 +42,10 @@ impl Annotations { fn has(&self, key: &str) -> bool { self.0.contains_key(key) } + + fn get(&self, key: &str) -> Option<&str> { + self.0.get(key).map(String::as_str) + } } impl From> for Annotations { @@ -122,6 +126,8 @@ impl Field { pub const KEY: &'static str = "key"; /// Annotation key marking a field as the watermark column for pagination. pub const CURSOR: &'static str = "cursor"; + /// Annotation key: `"target_entity.target_field"` this field references. + pub const REF: &'static str = "ref"; pub fn new(name: impl Into, data_type: DataType, nullable: bool) -> Self { Field { @@ -149,6 +155,10 @@ impl Field { self.annotations.has(Self::CURSOR) } + pub fn ref_target(&self) -> Option<&str> { + self.annotations.get(Self::REF) + } + pub fn with_description(mut self, description: impl Into) -> Self { self.description = Some(description.into()); self diff --git a/crates/strata/src/graphql/mod.rs b/crates/strata/src/graphql/mod.rs index da7a548..a001c69 100644 --- a/crates/strata/src/graphql/mod.rs +++ b/crates/strata/src/graphql/mod.rs @@ -10,6 +10,7 @@ //! come from the selection set and drive `?fields=` (projection). One page is //! served per query (the first chunk). +use std::collections::HashSet; use std::sync::Arc; use anyhow::Result; @@ -55,9 +56,7 @@ async fn graphiql() -> impl IntoResponse { /// Build the dynamic schema: one query field per queryable table across all mounts. /// Snapshotted at startup (tables added later need a rebuild). async fn build_schema(registry: &Arc) -> Result { - let mut query = Object::new("Query"); - let mut objects = Vec::new(); - + let mut tables_meta = Vec::new(); for mount in registry.names() { for table in tables(registry, &mount).await { let path = format!("/tables/{table}"); @@ -68,17 +67,30 @@ async fn build_schema(registry: &Arc) -> Result { if !endpoint.metadata.queryable { continue; } - let row_schema = endpoint.response; let type_name = format!("{mount}_{table}"); - objects.push(row_object(&type_name, &row_schema)); - query = query.field(table_field( - &type_name, - registry.clone(), - mount.clone(), - table, - )); + tables_meta.push((mount.clone(), table, type_name, endpoint.response)); } } + let known: HashSet = tables_meta.iter().map(|(_, _, t, _)| t.clone()).collect(); + + let mut query = Object::new("Query"); + let mut objects = Vec::new(); + for (mount, table, type_name, row_schema) in &tables_meta { + objects.push(row_object( + type_name, + row_schema, + registry.clone(), + mount.clone(), + &known, + )); + query = query.field(table_field( + type_name, + registry.clone(), + mount.clone(), + table.clone(), + row_schema, + )); + } let mut builder = Schema::build("Query", None, None) .register(Scalar::new(LONG)) @@ -112,7 +124,15 @@ async fn tables(registry: &Arc, mount: &str) -> Vec { /// A GraphQL object type for a row: one field per column, each reading its value /// out of the parent JSON row. All fields nullable (providers return nulls freely). -fn row_object(type_name: &str, row_schema: &StrataSchema) -> Object { +/// A column with a `ref` annotation also gets a nested field pointing at the +/// referenced table's type, resolved by a filtered read. +fn row_object( + type_name: &str, + row_schema: &StrataSchema, + registry: Arc, + mount: String, + known: &HashSet, +) -> Object { let mut object = Object::new(type_name); for field in &row_schema.fields { let key = field.name.clone(); @@ -132,25 +152,93 @@ fn row_object(type_name: &str, row_schema: &StrataSchema) -> Object { }) }, )); + + if let Some((target_table, target_col)) = field.ref_target().and_then(|t| t.split_once('.')) + && known.contains(&format!("{mount}_{target_table}")) + { + object = object.field(nested_field( + &field.name, + &format!("{mount}_{target_table}"), + target_table, + target_col, + registry.clone(), + mount.clone(), + )); + } } object } +fn nested_field( + column: &str, + target_type: &str, + target_table: &str, + target_col: &str, + registry: Arc, + mount: String, +) -> Field { + let name = nested_name(column); + let column = column.to_string(); + let target_table = target_table.to_string(); + let target_col = target_col.to_string(); + Field::new(name, TypeRef::named(target_type), move |ctx| { + let registry = registry.clone(); + let mount = mount.clone(); + let column = column.clone(); + let target_table = target_table.clone(); + let target_col = target_col.clone(); + FieldFuture::new(async move { + let row = ctx.parent_value.try_downcast_ref::()?; + let fk = match row.get(&column) { + Some(value) if !value.is_null() => value.clone(), + _ => return Ok(None), + }; + let filter = + serde_json::json!({ "cmp": { "field": target_col, "op": "eq", "value": fk } }); + let encoded = urlencoding::encode(&filter.to_string()).into_owned(); + let path = format!("/tables/{target_table}?filter={encoded}&limit=1"); + let rows = match registry.get(&mount)?.read(&path).await?.first().await? { + Some(chunk) => chunk.records.to_json_rows()?, + None => Vec::new(), + }; + Ok(rows.into_iter().next().map(FieldValue::owned_any)) + }) + }) +} + /// The `Query._` field: read one page, honoring `where` and `limit`, /// projecting to the selected columns. -fn table_field(type_name: &str, registry: Arc, mount: String, table: String) -> Field { +fn table_field( + type_name: &str, + registry: Arc, + mount: String, + table: String, + row_schema: &StrataSchema, +) -> Field { let field_name = type_name.to_string(); + let columns: HashSet = row_schema.fields.iter().map(|f| f.name.clone()).collect(); + let rel_to_fk: std::collections::HashMap = row_schema + .fields + .iter() + .filter(|f| f.ref_target().is_some()) + .map(|f| (nested_name(&f.name), f.name.clone())) + .collect(); Field::new(field_name, TypeRef::named_nn_list(type_name), move |ctx| { let registry = registry.clone(); let (mount, table) = (mount.clone(), table.clone()); + let (columns, rel_to_fk) = (columns.clone(), rel_to_fk.clone()); FieldFuture::new(async move { - // `?fields=` from the selection set (skip introspection meta fields). - let fields: Vec = ctx - .field() - .selection_set() - .map(|f| f.name().to_string()) - .filter(|n| !n.starts_with("__")) - .collect(); + let mut fields: Vec = Vec::new(); + for selected in ctx.field().selection_set() { + let name = selected.name(); + if columns.contains(name) { + fields.push(name.to_string()); + } else if let Some(fk) = rel_to_fk.get(name) + && !fields.contains(fk) + { + fields.push(fk.clone()); + } + } let filter = ctx .args @@ -174,6 +262,13 @@ fn table_field(type_name: &str, registry: Arc, mount: String, table: S .argument(InputValue::new("limit", TypeRef::named(TypeRef::INT))) } +fn nested_name(column: &str) -> String { + column + .strip_suffix("_id") + .filter(|s| !s.is_empty()) + .map_or_else(|| format!("{column}_ref"), String::from) +} + /// `/tables/
` with `filter`/`fields`/`limit` query params set. fn read_path(table: &str, filter: Option<&Value>, fields: &[String], limit: Option) -> String { let mut params: Vec<(String, String)> = Vec::new(); @@ -216,7 +311,7 @@ mod tests { use crate::pipe::{run_pass, store::NoPipeStore}; use crate::providers::dummy::Dummy; use crate::providers::sqlite::Sqlite; - use schema::{DataType, SchemaBuilder}; + use schema::{DataType, HasSchema, SchemaBuilder}; use serde_json::json; use strata_types::{Endpoint, Pipe}; @@ -285,4 +380,97 @@ mod tests { let _ = std::fs::remove_file(&db_path); Ok(()) } + + struct StaticSource(std::collections::HashMap); + + impl crate::router::SchemaSource for StaticSource { + fn schema( + &self, + path: String, + ) -> crate::router::BoxFuture<'static, Result>> { + let found = self.0.get(&path).cloned(); + Box::pin(async move { Ok(found) }) + } + } + + async fn put( + registry: &Registry, + path: &str, + rows: &[T], + ) -> Result<()> { + let body = crate::Body { + data: Some(crate::Dataset::of(rows)?.into_stream()), + meta: Value::Null, + }; + registry + .get("local")? + .invoke(crate::Method::Put, path, Some(body)) + .await?; + Ok(()) + } + + #[tokio::test] + async fn resolves_a_declared_relationship() -> Result<()> { + #[derive(serde::Serialize, schema::HasSchema)] + struct Customer { + #[schema(key)] + id: i64, + name: String, + } + #[derive(serde::Serialize, schema::HasSchema)] + struct Order { + #[schema(key)] + id: i64, + customer_id: i64, + } + #[derive(serde::Serialize, schema::HasSchema)] + struct OrderRel { + #[schema(key)] + id: i64, + #[schema(references = "customers.id")] + customer_id: i64, + } + + let db_path = std::env::temp_dir().join("strata_graphql_rel.sqlite"); + let _ = std::fs::remove_file(&db_path); + + let mut registry = Registry::new(); + let cfg: ProviderConfig = serde_json::from_value( + json!({ "backend": "sqlite", "path": db_path.to_str().unwrap() }), + )?; + registry.mount::("local", &cfg)?; + + put( + ®istry, + "/tables/customers", + &[Customer { id: 1, name: "acme".into() }], + ) + .await?; + put( + ®istry, + "/tables/orders", + &[Order { id: 10, customer_id: 1 }], + ) + .await?; + + let mut declared = std::collections::HashMap::new(); + declared.insert("/tables/orders".to_string(), OrderRel::schema()); + registry.set_schema_source("local", Arc::new(StaticSource(declared)))?; + + let registry = Arc::new(registry); + let schema = build_schema(®istry).await?; + + let response = schema + .execute("{ local_orders { id customer { id name } } }") + .await; + assert!(response.errors.is_empty(), "graphql errors: {:?}", response.errors); + let data = response.data.into_json().unwrap(); + let order = &data["local_orders"][0]; + assert_eq!(order["id"], 10); + assert_eq!(order["customer"]["id"], 1); + assert_eq!(order["customer"]["name"], "acme"); + + let _ = std::fs::remove_file(&db_path); + Ok(()) + } } diff --git a/crates/strata/src/provider.rs b/crates/strata/src/provider.rs index f209f09..6ed0173 100644 --- a/crates/strata/src/provider.rs +++ b/crates/strata/src/provider.rs @@ -14,7 +14,7 @@ use serde_json::{Value, json}; use crate::catalog::Catalog; use crate::config::ProviderConfig; use crate::dataset::DataStream; -use crate::router::{Body, BoxFuture, EndpointInfo, Method, Response, Router}; +use crate::router::{Body, BoxFuture, EndpointInfo, Method, Response, Router, SchemaSource}; /// Implemented by each concrete provider. Knows its state type and how to wire /// its routes. This is the typed, ergonomic surface provider authors write. @@ -36,9 +36,7 @@ pub trait Provider: Sized + Send + Sync + 'static { /// being an `async fn`. pub trait ProviderObject: Send + Sync { fn name(&self) -> &str; - /// Wire the mount-scoped [`Catalog`] into this provider's router, so its - /// handlers can read their own persisted annotations. - fn set_catalog(&mut self, catalog: Catalog); + fn set_schema_source(&mut self, source: Arc); /// Every endpoint, statically described (dynamic resolvers not run). fn endpoints(&self) -> Vec; /// The read endpoint matching a concrete `path`, with its response schema @@ -73,8 +71,8 @@ where S::name() } - fn set_catalog(&mut self, catalog: Catalog) { - self.router.set_catalog(catalog); + fn set_schema_source(&mut self, source: Arc) { + self.router.set_schema_source(source); } fn endpoints(&self) -> Vec { @@ -136,10 +134,18 @@ impl Registry { pub fn set_catalog(&mut self, db: database::Database) { for (mount, provider) in self.providers.iter_mut() { - provider.set_catalog(Catalog::new(db.clone(), mount.clone())); + provider.set_schema_source(Arc::new(Catalog::new(db.clone(), mount.clone()))); } } + pub fn set_schema_source(&mut self, mount: &str, source: Arc) -> Result<()> { + self.providers + .get_mut(mount) + .ok_or_else(|| anyhow!("nothing mounted at `{mount}`"))? + .set_schema_source(source); + Ok(()) + } + /// Look up a mounted provider by its mount point. pub fn get(&self, mount: &str) -> Result<&dyn ProviderObject> { self.providers.get(mount).map(Box::as_ref).ok_or_else(|| { diff --git a/crates/strata/src/providers/sql/mod.rs b/crates/strata/src/providers/sql/mod.rs index c220ea0..7689fe5 100644 --- a/crates/strata/src/providers/sql/mod.rs +++ b/crates/strata/src/providers/sql/mod.rs @@ -349,7 +349,7 @@ pub async fn table_data(db: Arc, p: Params) -> Result Option { /// The `.data_type()` resolver for `/tables/:table/data`. pub async fn table_data_schema(db: Arc, p: Params) -> Result { - let full = db.table_schema(p.get("table")?).await?; + let full = enrich(db.table_schema(p.get("table")?).await?, p.schema()); project_schema(&full, get_projection(&p).as_deref()) } +fn enrich(mut schema: Schema, annotated: Option<&Schema>) -> Schema { + let Some(annotated) = annotated else { + return schema; + }; + for field in &mut schema.fields { + if let Some(source) = annotated.fields.iter().find(|f| f.name == field.name) { + for (key, value) in source.annotations.to_map() { + field.annotate(key, value); + } + } + } + schema +} + /// `put /tables/:table`: sink a [`DataStream`]. Decodes the `disposition` param and /// validates the shared write contract off the stream's schema (row-shaped, /// non-empty; a key when merging), then hands the stream to the provider. diff --git a/crates/strata/src/router.rs b/crates/strata/src/router.rs index 0f731af..a9a2cdd 100644 --- a/crates/strata/src/router.rs +++ b/crates/strata/src/router.rs @@ -74,6 +74,10 @@ impl Params { self.query.get(key).map(String::as_str) } + pub fn schema(&self) -> Option<&Schema> { + self.schema.as_ref() + } + /// The resume cursor, decoded into the provider's own cursor state `C` — the /// inverse of [`Cursor::new`]. On the first page (no `cursor` param) it decodes /// `{}`, so `C`'s `#[serde(default)]` fields supply the starting position. @@ -703,7 +707,7 @@ impl Serialize for EndpointInfo { /// Maps route patterns to handlers for one provider whose state is `S`. pub struct Router { routes: Vec>, - schema_source: Option>, + schema_source: Option>, } impl Default for Router { @@ -722,8 +726,8 @@ impl Router { /// Wire the schema source (called by `serve` once the catalog DB is up), so /// dispatch can hand each handler the persisted schema for its endpoint. - pub fn set_catalog(&mut self, source: impl SchemaSource + 'static) { - self.schema_source = Some(Box::new(source)); + pub fn set_schema_source(&mut self, source: Arc) { + self.schema_source = Some(source); } /// Register a built [`Route`]. Panics at startup if the route has no diff --git a/crates/strata/src/testkit.rs b/crates/strata/src/testkit.rs index dae46a7..a26550c 100644 --- a/crates/strata/src/testkit.rs +++ b/crates/strata/src/testkit.rs @@ -41,7 +41,8 @@ impl Client { /// SQL read derives its cursor column from the schema's `cursor` annotation. pub fn with_schema(mut self, path: &str, schema: Schema) -> Self { self.schemas.insert(path.to_string(), schema); - self.router.set_catalog(StaticSchemas(self.schemas.clone())); + self.router + .set_schema_source(Arc::new(StaticSchemas(self.schemas.clone()))); self }