1use crate::algebra::{Expression, Term, Variable};
7use anyhow::{anyhow, bail, Result};
8use oxirs_core::model::NamedNode;
9use std::any::Any;
10use std::collections::HashMap;
11use std::fmt::Debug;
12use std::sync::{Arc, RwLock};
13
14#[derive(Debug)]
16pub struct ExtensionRegistry {
17 pub functions: Arc<RwLock<HashMap<String, Box<dyn CustomFunction>>>>,
19 pub operators: Arc<RwLock<HashMap<String, Box<dyn CustomOperator>>>>,
21 pub aggregates: Arc<RwLock<HashMap<String, Box<dyn CustomAggregate>>>>,
23 pub plugins: Arc<RwLock<Vec<Box<dyn ExtensionPlugin>>>>,
25 pub type_converters: Arc<RwLock<HashMap<String, Box<dyn TypeConverter>>>>,
27}
28
29pub trait CustomFunction: Send + Sync + Debug {
31 fn name(&self) -> &str;
33
34 fn arity(&self) -> Option<usize>;
36
37 fn parameter_types(&self) -> Vec<ValueType>;
39
40 fn return_type(&self) -> ValueType;
42
43 fn documentation(&self) -> &str;
45
46 fn execute(&self, args: &[Value], context: &ExecutionContext) -> Result<Value>;
48
49 fn clone_function(&self) -> Box<dyn CustomFunction>;
51
52 fn validate(&self, args: &[Expression]) -> Result<()> {
54 if let Some(expected_arity) = self.arity() {
55 if args.len() != expected_arity {
56 bail!(
57 "Function {} expects {} arguments, got {}",
58 self.name(),
59 expected_arity,
60 args.len()
61 );
62 }
63 }
64 Ok(())
65 }
66
67 fn cost_estimate(&self, args: &[Expression]) -> f64 {
69 100.0 + args.len() as f64 * 10.0
71 }
72
73 fn is_deterministic(&self) -> bool {
75 true
76 }
77
78 fn can_pushdown(&self) -> bool {
80 self.is_deterministic()
81 }
82}
83
84pub trait CustomOperator: Send + Sync + Debug {
86 fn symbol(&self) -> &str;
88
89 fn precedence(&self) -> i32;
91
92 fn associativity(&self) -> Associativity;
94
95 fn operator_type(&self) -> OperatorType;
97
98 fn execute(
100 &self,
101 left: Option<&Value>,
102 right: Option<&Value>,
103 context: &ExecutionContext,
104 ) -> Result<Value>;
105
106 fn type_check(
108 &self,
109 left_type: Option<ValueType>,
110 right_type: Option<ValueType>,
111 ) -> Result<ValueType>;
112}
113
114pub trait CustomAggregate: Send + Sync + Debug {
116 fn name(&self) -> &str;
118
119 fn init(&self) -> Box<dyn AggregateState>;
121
122 fn supports_distinct(&self) -> bool {
124 true
125 }
126
127 fn documentation(&self) -> &str;
129}
130
131pub trait AggregateState: Send + Sync + Debug {
133 fn add(&mut self, value: &Value) -> Result<()>;
135
136 fn result(&self) -> Result<Value>;
138
139 fn reset(&mut self);
141
142 fn clone_state(&self) -> Box<dyn AggregateState>;
144}
145
146pub trait ExtensionPlugin: Send + Sync + Debug {
148 fn name(&self) -> &str;
150
151 fn version(&self) -> &str;
153
154 fn dependencies(&self) -> Vec<String>;
156
157 fn initialize(&mut self, registry: &mut ExtensionRegistry) -> Result<()>;
159
160 fn shutdown(&mut self) -> Result<()>;
162
163 fn metadata(&self) -> PluginMetadata;
165}
166
167#[derive(Debug, Clone)]
169pub struct PluginMetadata {
170 pub name: String,
171 pub version: String,
172 pub author: String,
173 pub description: String,
174 pub license: String,
175 pub homepage: Option<String>,
176 pub repository: Option<String>,
177}
178
179pub trait TypeConverter: Send + Sync + Debug {
181 #[allow(clippy::wrong_self_convention)]
183 fn from_type(&self) -> &str;
184
185 fn to_type(&self) -> &str;
187
188 fn convert(&self, value: &Value) -> Result<Value>;
190
191 fn can_convert(&self, value: &Value) -> bool;
193}
194
195#[derive(Debug, Clone, PartialEq)]
197pub enum ValueType {
198 String,
199 Integer,
200 Float,
201 Boolean,
202 DateTime,
203 Duration,
204 Iri,
205 BlankNode,
206 Literal,
207 Custom(String),
208 List(Box<ValueType>),
209 Optional(Box<ValueType>),
210 Union(Vec<ValueType>),
211}
212
213#[derive(Debug)]
215pub enum Value {
216 String(String),
217 Integer(i64),
218 Float(f64),
219 Boolean(bool),
220 DateTime(chrono::DateTime<chrono::Utc>),
221 Duration(chrono::Duration),
222 Iri(String),
223 BlankNode(String),
224 Literal {
225 value: String,
226 language: Option<String>,
227 datatype: Option<String>,
228 },
229 List(Vec<Value>),
230 Null,
231 Custom {
232 type_name: String,
233 data: Box<dyn Any + Send + Sync>,
234 },
235}
236
237impl Clone for Value {
238 fn clone(&self) -> Self {
239 match self {
240 Value::String(s) => Value::String(s.clone()),
241 Value::Integer(i) => Value::Integer(*i),
242 Value::Float(f) => Value::Float(*f),
243 Value::Boolean(b) => Value::Boolean(*b),
244 Value::DateTime(dt) => Value::DateTime(*dt),
245 Value::Duration(d) => Value::Duration(*d),
246 Value::Iri(iri) => Value::Iri(iri.clone()),
247 Value::BlankNode(id) => Value::BlankNode(id.clone()),
248 Value::Literal {
249 value,
250 language,
251 datatype,
252 } => Value::Literal {
253 value: value.clone(),
254 language: language.clone(),
255 datatype: datatype.clone(),
256 },
257 Value::List(list) => Value::List(list.clone()),
258 Value::Null => Value::Null,
259 Value::Custom { type_name, .. } => {
260 Value::String(format!("Custom({type_name})"))
262 }
263 }
264 }
265}
266
267impl PartialEq for Value {
268 fn eq(&self, other: &Self) -> bool {
269 match (self, other) {
270 (Value::String(a), Value::String(b)) => a == b,
271 (Value::Integer(a), Value::Integer(b)) => a == b,
272 (Value::Float(a), Value::Float(b)) => a == b,
273 (Value::Boolean(a), Value::Boolean(b)) => a == b,
274 (Value::DateTime(a), Value::DateTime(b)) => a == b,
275 (Value::Duration(a), Value::Duration(b)) => a == b,
276 (Value::Iri(a), Value::Iri(b)) => a == b,
277 (Value::BlankNode(a), Value::BlankNode(b)) => a == b,
278 (
279 Value::Literal {
280 value: v1,
281 language: l1,
282 datatype: d1,
283 },
284 Value::Literal {
285 value: v2,
286 language: l2,
287 datatype: d2,
288 },
289 ) => v1 == v2 && l1 == l2 && d1 == d2,
290 (Value::List(a), Value::List(b)) => a == b,
291 (Value::Null, Value::Null) => true,
292 (Value::Custom { type_name: t1, .. }, Value::Custom { type_name: t2, .. }) => t1 == t2,
293 _ => false,
294 }
295 }
296}
297
298impl PartialOrd for Value {
299 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
300 use std::cmp::Ordering;
301 match (self, other) {
302 (Value::String(a), Value::String(b)) => a.partial_cmp(b),
303 (Value::Integer(a), Value::Integer(b)) => a.partial_cmp(b),
304 (Value::Float(a), Value::Float(b)) => a.partial_cmp(b),
305 (Value::Boolean(a), Value::Boolean(b)) => a.partial_cmp(b),
306 (Value::DateTime(a), Value::DateTime(b)) => a.partial_cmp(b),
307 (Value::Duration(a), Value::Duration(b)) => a.partial_cmp(b),
308 (Value::Iri(a), Value::Iri(b)) => a.partial_cmp(b),
309 (Value::BlankNode(a), Value::BlankNode(b)) => a.partial_cmp(b),
310 (
311 Value::Literal {
312 value: v1,
313 language: l1,
314 datatype: d1,
315 },
316 Value::Literal {
317 value: v2,
318 language: l2,
319 datatype: d2,
320 },
321 ) => match v1.partial_cmp(v2) {
322 Some(Ordering::Equal) => match l1.partial_cmp(l2) {
323 Some(Ordering::Equal) => d1.partial_cmp(d2),
324 other => other,
325 },
326 other => other,
327 },
328 (Value::Integer(a), Value::Float(b)) => (*a as f64).partial_cmp(b),
329 (Value::Float(a), Value::Integer(b)) => a.partial_cmp(&(*b as f64)),
330 (Value::Null, Value::Null) => Some(Ordering::Equal),
331 (Value::Null, _) => Some(Ordering::Less),
332 (_, Value::Null) => Some(Ordering::Greater),
333 _ => None, }
335 }
336}
337
338#[derive(Debug, Clone, PartialEq)]
340pub enum Associativity {
341 Left,
342 Right,
343 None,
344}
345
346#[derive(Debug, Clone, PartialEq)]
348pub enum OperatorType {
349 Binary,
350 Unary,
351 Ternary,
352}
353
354#[derive(Debug, Clone)]
356pub struct ExecutionContext {
357 pub variables: HashMap<Variable, Term>,
358 pub namespaces: HashMap<String, String>,
359 pub base_iri: Option<String>,
360 pub dataset_context: Option<String>,
361 pub query_time: chrono::DateTime<chrono::Utc>,
362 pub optimization_level: OptimizationLevel,
363 pub memory_limit: Option<usize>,
364 pub time_limit: Option<std::time::Duration>,
365}
366
367impl Default for ExecutionContext {
368 fn default() -> Self {
369 Self {
370 variables: HashMap::new(),
371 namespaces: HashMap::new(),
372 base_iri: None,
373 dataset_context: None,
374 query_time: chrono::Utc::now(),
375 optimization_level: OptimizationLevel::None,
376 memory_limit: None,
377 time_limit: None,
378 }
379 }
380}
381
382#[derive(Debug, Clone, PartialEq)]
384pub enum OptimizationLevel {
385 None,
386 Basic,
387 Aggressive,
388}
389
390impl ExtensionRegistry {
391 pub fn new() -> Self {
392 Self {
393 functions: Arc::new(RwLock::new(HashMap::new())),
394 operators: Arc::new(RwLock::new(HashMap::new())),
395 aggregates: Arc::new(RwLock::new(HashMap::new())),
396 plugins: Arc::new(RwLock::new(Vec::new())),
397 type_converters: Arc::new(RwLock::new(HashMap::new())),
398 }
399 }
400
401 pub fn register_function<F>(&self, function: F) -> Result<()>
403 where
404 F: CustomFunction + 'static,
405 {
406 let name = function.name().to_string();
407 let mut functions = self
408 .functions
409 .write()
410 .map_err(|_| anyhow!("Failed to acquire write lock on functions"))?;
411 functions.insert(name, Box::new(function));
412 Ok(())
413 }
414
415 pub fn register_operator<O>(&self, operator: O) -> Result<()>
417 where
418 O: CustomOperator + 'static,
419 {
420 let symbol = operator.symbol().to_string();
421 let mut operators = self
422 .operators
423 .write()
424 .map_err(|_| anyhow!("Failed to acquire write lock on operators"))?;
425 operators.insert(symbol, Box::new(operator));
426 Ok(())
427 }
428
429 pub fn register_aggregate<A>(&self, aggregate: A) -> Result<()>
431 where
432 A: CustomAggregate + 'static,
433 {
434 let name = aggregate.name().to_string();
435 let mut aggregates = self
436 .aggregates
437 .write()
438 .map_err(|_| anyhow!("Failed to acquire write lock on aggregates"))?;
439 aggregates.insert(name, Box::new(aggregate));
440 Ok(())
441 }
442
443 pub fn register_plugin<P>(&mut self, mut plugin: P) -> Result<()>
445 where
446 P: ExtensionPlugin + 'static,
447 {
448 plugin.initialize(self)?;
450
451 let mut plugins = self
452 .plugins
453 .write()
454 .map_err(|_| anyhow!("Failed to acquire write lock on plugins"))?;
455 plugins.push(Box::new(plugin));
456 Ok(())
457 }
458
459 pub fn register_type_converter<T>(&self, converter: T) -> Result<()>
461 where
462 T: TypeConverter + 'static,
463 {
464 let key = format!("{}:{}", converter.from_type(), converter.to_type());
465 let mut converters = self
466 .type_converters
467 .write()
468 .map_err(|_| anyhow!("Failed to acquire write lock on type converters"))?;
469 converters.insert(key, Box::new(converter));
470 Ok(())
471 }
472
473 pub fn get_function(&self, name: &str) -> Result<Option<Box<dyn CustomFunction>>> {
475 let functions = self
476 .functions
477 .read()
478 .map_err(|_| anyhow!("Failed to acquire read lock on functions"))?;
479 Ok(functions.get(name).map(|f| f.clone_function()))
480 }
481
482 pub fn has_function(&self, name: &str) -> Result<bool> {
484 let functions = self
485 .functions
486 .read()
487 .map_err(|_| anyhow!("Failed to acquire read lock on functions"))?;
488 Ok(functions.contains_key(name))
489 }
490
491 pub fn has_operator(&self, symbol: &str) -> Result<bool> {
493 let operators = self
494 .operators
495 .read()
496 .map_err(|_| anyhow!("Failed to acquire read lock on operators"))?;
497 Ok(operators.contains_key(symbol))
498 }
499
500 pub fn has_aggregate(&self, name: &str) -> Result<bool> {
502 let aggregates = self
503 .aggregates
504 .read()
505 .map_err(|_| anyhow!("Failed to acquire read lock on aggregates"))?;
506 Ok(aggregates.contains_key(name))
507 }
508
509 pub fn execute_function(
511 &self,
512 name: &str,
513 args: &[Value],
514 context: &ExecutionContext,
515 ) -> Result<Value> {
516 let functions = self
517 .functions
518 .read()
519 .map_err(|_| anyhow!("Failed to acquire read lock on functions"))?;
520
521 if let Some(func) = functions.get(name) {
522 func.execute(args, context)
523 } else {
524 Err(anyhow!("Function '{}' not found", name))
525 }
526 }
527
528 pub fn execute_operator(
530 &self,
531 symbol: &str,
532 left: Option<&Value>,
533 right: Option<&Value>,
534 context: &ExecutionContext,
535 ) -> Result<Value> {
536 let operators = self
537 .operators
538 .read()
539 .map_err(|_| anyhow!("Failed to acquire read lock on operators"))?;
540
541 if let Some(op) = operators.get(symbol) {
542 op.execute(left, right, context)
543 } else {
544 Err(anyhow!("Operator '{}' not found", symbol))
545 }
546 }
547
548 pub fn create_aggregate_state(&self, name: &str) -> Result<Box<dyn AggregateState>> {
550 let aggregates = self
551 .aggregates
552 .read()
553 .map_err(|_| anyhow!("Failed to acquire read lock on aggregates"))?;
554
555 if let Some(agg) = aggregates.get(name) {
556 Ok(agg.init())
557 } else {
558 Err(anyhow!("Aggregate '{}' not found", name))
559 }
560 }
561
562 pub fn convert_value(&self, value: &Value, target_type: &str) -> Result<Value> {
564 let source_type = value.type_name();
565 let key = format!("{source_type}:{target_type}");
566
567 let converters = self
568 .type_converters
569 .read()
570 .map_err(|_| anyhow!("Failed to acquire read lock on type converters"))?;
571
572 if let Some(converter) = converters.get(&key) {
573 converter.convert(value)
574 } else {
575 self.builtin_convert(value, target_type)
577 }
578 }
579
580 fn builtin_convert(&self, value: &Value, target_type: &str) -> Result<Value> {
582 match (value, target_type) {
583 (Value::String(s), "integer") => s
584 .parse::<i64>()
585 .map(Value::Integer)
586 .map_err(|_| anyhow!("Cannot convert '{}' to integer", s)),
587 (Value::String(s), "float") => s
588 .parse::<f64>()
589 .map(Value::Float)
590 .map_err(|_| anyhow!("Cannot convert '{}' to float", s)),
591 (Value::Integer(i), "string") => Ok(Value::String(i.to_string())),
592 (Value::Float(f), "string") => Ok(Value::String(f.to_string())),
593 (Value::Boolean(b), "string") => Ok(Value::String(b.to_string())),
594 _ => bail!(
595 "No conversion available from {} to {}",
596 value.type_name(),
597 target_type
598 ),
599 }
600 }
601
602 pub fn list_functions(&self) -> Result<Vec<String>> {
604 let functions = self
605 .functions
606 .read()
607 .map_err(|_| anyhow!("Failed to acquire read lock on functions"))?;
608 Ok(functions.keys().cloned().collect())
609 }
610
611 pub fn list_operators(&self) -> Result<Vec<String>> {
613 let operators = self
614 .operators
615 .read()
616 .map_err(|_| anyhow!("Failed to acquire read lock on operators"))?;
617 Ok(operators.keys().cloned().collect())
618 }
619
620 pub fn validate_extensions(&self) -> Result<Vec<String>> {
622 let mut errors = Vec::new();
623
624 let plugins = self
626 .plugins
627 .read()
628 .map_err(|_| anyhow!("Failed to acquire read lock on plugins"))?;
629
630 for plugin in plugins.iter() {
631 for dep in plugin.dependencies() {
632 let found = plugins.iter().any(|p| p.name() == dep);
633 if !found {
634 errors.push(format!(
635 "Plugin '{}' missing dependency '{}'",
636 plugin.name(),
637 dep
638 ));
639 }
640 }
641 }
642
643 Ok(errors)
644 }
645}
646
647impl Default for ExtensionRegistry {
648 fn default() -> Self {
649 Self::new()
650 }
651}
652
653impl Value {
654 pub fn type_name(&self) -> &str {
656 match self {
657 Value::String(_) => "string",
658 Value::Integer(_) => "integer",
659 Value::Float(_) => "float",
660 Value::Boolean(_) => "boolean",
661 Value::DateTime(_) => "datetime",
662 Value::Duration(_) => "duration",
663 Value::Iri(_) => "iri",
664 Value::BlankNode(_) => "bnode",
665 Value::Literal { .. } => "literal",
666 Value::List(_) => "list",
667 Value::Null => "null",
668 Value::Custom { type_name, .. } => type_name,
669 }
670 }
671
672 pub fn to_term(&self) -> Result<Term> {
674 match self {
675 Value::String(s) => Ok(Term::Literal(crate::algebra::Literal {
676 value: s.clone(),
677 language: None,
678 datatype: None,
679 })),
680 Value::Iri(iri) => Ok(Term::Iri(NamedNode::new_unchecked(iri.clone()))),
681 Value::BlankNode(id) => Ok(Term::BlankNode(id.clone())),
682 Value::Literal {
683 value,
684 language,
685 datatype,
686 } => Ok(Term::Literal(crate::algebra::Literal {
687 value: value.clone(),
688 language: language.clone(),
689 datatype: datatype
690 .as_ref()
691 .map(|dt| NamedNode::new_unchecked(dt.clone())),
692 })),
693 _ => bail!("Cannot convert {} to Term", self.type_name()),
694 }
695 }
696
697 pub fn from_term(term: &Term) -> Self {
699 match term {
700 Term::Iri(iri) => Value::Iri(iri.as_str().to_string()),
701 Term::BlankNode(id) => Value::BlankNode(id.clone()),
702 Term::Literal(lit) => Value::Literal {
703 value: lit.value.clone(),
704 language: lit.language.clone(),
705 datatype: lit.datatype.as_ref().map(|dt| dt.as_str().to_string()),
706 },
707 Term::Variable(var) => Value::String(format!("?{var}")),
708 Term::QuotedTriple(_) => Value::String("<<quoted triple>>".to_string()),
709 Term::PropertyPath(_) => Value::String("<property path>".to_string()),
710 }
711 }
712}
713
714#[macro_export]
716macro_rules! register_function {
717 ($registry:expr_2021, $name:expr_2021, $params:expr_2021, $return_type:expr_2021, $body:expr_2021) => {{
718 #[derive(Debug, Clone)]
719 struct GeneratedFunction {
720 name: String,
721 params: Vec<ValueType>,
722 return_type: ValueType,
723 body: fn(&[Value], &ExecutionContext) -> Result<Value>,
724 }
725
726 impl CustomFunction for GeneratedFunction {
727 fn name(&self) -> &str {
728 &self.name
729 }
730 fn arity(&self) -> Option<usize> {
731 Some(self.params.len())
732 }
733 fn parameter_types(&self) -> Vec<ValueType> {
734 self.params.clone()
735 }
736 fn return_type(&self) -> ValueType {
737 self.return_type.clone()
738 }
739 fn documentation(&self) -> &str {
740 "Generated function"
741 }
742 fn clone_function(&self) -> Box<dyn CustomFunction> {
743 Box::new(self.clone())
744 }
745
746 fn execute(&self, args: &[Value], context: &ExecutionContext) -> Result<Value> {
747 (self.body)(args, context)
748 }
749 }
750
751 let func = GeneratedFunction {
752 name: $name.to_string(),
753 params: $params,
754 return_type: $return_type,
755 body: $body,
756 };
757
758 $registry.register_function(func)
759 }};
760}
761
762#[cfg(test)]
763mod tests {
764 use super::*;
765
766 #[derive(Debug, Clone)]
767 struct TestFunction;
768
769 impl CustomFunction for TestFunction {
770 fn name(&self) -> &str {
771 "http://example.org/test"
772 }
773 fn arity(&self) -> Option<usize> {
774 Some(2)
775 }
776 fn parameter_types(&self) -> Vec<ValueType> {
777 vec![ValueType::Integer, ValueType::Integer]
778 }
779 fn return_type(&self) -> ValueType {
780 ValueType::Integer
781 }
782 fn documentation(&self) -> &str {
783 "Test function that adds two integers"
784 }
785 fn clone_function(&self) -> Box<dyn CustomFunction> {
786 Box::new(self.clone())
787 }
788
789 fn execute(&self, args: &[Value], _context: &ExecutionContext) -> Result<Value> {
790 if args.len() != 2 {
791 bail!("Expected 2 arguments, got {}", args.len());
792 }
793
794 match (&args[0], &args[1]) {
795 (Value::Integer(a), Value::Integer(b)) => Ok(Value::Integer(a + b)),
796 _ => bail!("Expected integer arguments"),
797 }
798 }
799 }
800
801 #[test]
802 fn test_function_registration() {
803 let registry = ExtensionRegistry::new();
804 let func = TestFunction;
805
806 assert!(registry.register_function(func).is_ok());
807 assert!(registry
808 .get_function("http://example.org/test")
809 .unwrap()
810 .is_some());
811 }
812
813 #[test]
814 fn test_function_execution() {
815 let func = TestFunction;
816 let args = vec![Value::Integer(5), Value::Integer(3)];
817 let context = ExecutionContext {
818 variables: HashMap::new(),
819 namespaces: HashMap::new(),
820 base_iri: None,
821 dataset_context: None,
822 query_time: chrono::Utc::now(),
823 optimization_level: OptimizationLevel::Basic,
824 memory_limit: None,
825 time_limit: None,
826 };
827
828 let result = func.execute(&args, &context).unwrap();
829 assert_eq!(result, Value::Integer(8));
830 }
831}