1use std::collections::HashMap;
84use std::fmt;
85use std::sync::Arc;
86
87use substrait::proto::NamedStruct;
88use substrait::proto::r#type::{Nullability, Struct};
89use thiserror::Error;
90
91use crate::extensions::any::{Any, AnyRef};
92use crate::extensions::args::{ExtensionArgs, ExtensionColumn, ExtensionValueKind};
93
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
96pub enum ExtensionType {
97 Relation,
99 Enhancement,
101 Optimization,
103}
104
105#[derive(Debug, Error, Clone)]
107pub enum RegistrationError {
108 #[error("{ext_type:?} extension '{name}' already registered")]
109 DuplicateName {
110 ext_type: ExtensionType,
111 name: String,
112 },
113
114 #[error("Type URL '{type_url}' already registered to {ext_type:?} extension '{existing_name}'")]
115 ConflictingTypeUrl {
116 type_url: String,
117 ext_type: ExtensionType,
118 existing_name: String,
119 },
120}
121
122#[derive(Debug, Error, Clone)]
124pub enum ExtensionError {
125 #[error("Extension '{name}' not found in registry")]
127 NotFound { name: String },
128
129 #[error("Missing required argument: {name}")]
131 MissingArgument { name: String },
132
133 #[error("Invalid argument: expected {expected}, got {actual}")]
135 InvalidArgumentType {
136 expected: ExtensionValueKind,
137 actual: ExtensionValueKind,
138 },
139
140 #[error("Invalid argument: {0}")]
146 InvalidArgument(String),
147
148 #[error("Type URL mismatch: expected {expected}, got {actual}")]
150 TypeUrlMismatch { expected: String, actual: String },
151
152 #[error("Failed to decode protobuf message")]
154 DecodeFailed(#[source] prost::DecodeError),
155
156 #[error("Failed to encode protobuf message")]
158 EncodeFailed(#[source] prost::EncodeError),
159
160 #[error("Extension detail is missing")]
162 MissingDetail,
163
164 #[error("{0}")]
166 Custom(String),
167}
168
169pub trait AnyConvertible: Sized {
173 fn to_any(&self) -> Result<Any, ExtensionError>;
175
176 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError>;
178
179 fn type_url() -> String;
183}
184
185impl<T> AnyConvertible for T
187where
188 T: prost::Message + prost::Name + Default,
189{
190 fn to_any(&self) -> Result<Any, ExtensionError> {
191 Any::encode(self)
192 }
193
194 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
195 any.decode()
196 }
197
198 fn type_url() -> String {
199 T::type_url()
200 }
201}
202
203pub trait ExtensionProtoConvert<T> {
214 fn convert(&self) -> Result<T, ExtensionError>;
216}
217
218impl ExtensionProtoConvert<NamedStruct> for [ExtensionColumn] {
219 fn convert(&self) -> Result<NamedStruct, ExtensionError> {
220 let mut names = Vec::with_capacity(self.len());
221 let mut types = Vec::with_capacity(self.len());
222 for col in self {
223 match col {
224 ExtensionColumn::Named { name, r#type: ty } => {
225 names.push(name.clone());
226 types.push(ty.clone());
227 }
228 other => {
229 return Err(ExtensionError::InvalidArgument(format!(
230 "Expected named column, got {other:?}"
231 )));
232 }
233 }
234 }
235 Ok(NamedStruct {
236 names,
237 r#struct: Some(Struct {
238 types,
239 type_variation_reference: 0,
240 nullability: Nullability::Required as i32,
244 }),
245 })
246 }
247}
248
249impl ExtensionProtoConvert<Vec<ExtensionColumn>> for NamedStruct {
250 fn convert(&self) -> Result<Vec<ExtensionColumn>, ExtensionError> {
251 let types = self
252 .r#struct
253 .as_ref()
254 .map(|s| s.types.as_slice())
255 .unwrap_or_default();
256 if self.names.len() != types.len() {
257 return Err(ExtensionError::InvalidArgument(format!(
258 "NamedStruct has {} names but {} types",
259 self.names.len(),
260 types.len()
261 )));
262 }
263 Ok(self
264 .names
265 .iter()
266 .zip(types.iter())
267 .map(|(name, ty)| ExtensionColumn::Named {
268 name: name.clone(),
269 r#type: ty.clone(),
270 })
271 .collect())
272 }
273}
274
275pub trait Explainable: Sized {
277 fn name() -> &'static str;
280
281 fn from_args(args: &ExtensionArgs) -> Result<Self, ExtensionError>;
283
284 fn to_args(&self) -> Result<ExtensionArgs, ExtensionError>;
286}
287
288trait ExtensionConverter: Send + Sync {
302 fn parse_detail(&self, args: &ExtensionArgs) -> Result<Any, ExtensionError>;
303
304 fn textify_detail(&self, detail: AnyRef<'_>) -> Result<ExtensionArgs, ExtensionError>;
305}
306
307struct ExtensionAdapter<T>(std::marker::PhantomData<T>);
324
325impl<T: AnyConvertible + Explainable + Send + Sync> ExtensionConverter for ExtensionAdapter<T> {
326 fn parse_detail(&self, args: &ExtensionArgs) -> Result<Any, ExtensionError> {
327 T::from_args(args)?.to_any()
328 }
329
330 fn textify_detail(&self, detail: AnyRef<'_>) -> Result<ExtensionArgs, ExtensionError> {
331 let owned_any = Any::new(detail.type_url.to_string(), detail.value.to_vec());
332 T::from_any(owned_any.as_ref())?.to_args()
333 }
334}
335
336pub trait Extension: AnyConvertible + Explainable + Send + Sync + 'static {}
337
338impl<T> Extension for T where T: AnyConvertible + Explainable + Send + Sync + 'static {}
339
340#[derive(Default, Clone)]
342pub struct ExtensionRegistry {
343 handlers: HashMap<(ExtensionType, String), Arc<dyn ExtensionConverter>>,
345 type_urls: HashMap<(ExtensionType, String), String>,
347 descriptors: Vec<Vec<u8>>,
353}
354
355impl ExtensionRegistry {
356 pub fn new() -> Self {
358 Self {
359 handlers: HashMap::new(),
360 type_urls: HashMap::new(),
361 descriptors: Vec::new(),
362 }
363 }
364
365 pub fn add_descriptor(&mut self, bytes: Vec<u8>) {
373 self.descriptors.push(bytes);
374 }
375
376 pub fn descriptors(&self) -> Vec<&[u8]> {
378 self.descriptors.iter().map(|b| b.as_slice()).collect()
379 }
380
381 fn register<T>(&mut self, ext_type: ExtensionType) -> Result<(), RegistrationError>
383 where
384 T: Extension,
385 {
386 let canonical_name = T::name();
387 let type_url = T::type_url();
388 let handler: Arc<dyn ExtensionConverter> =
389 Arc::new(ExtensionAdapter::<T>(std::marker::PhantomData));
390
391 let key = (ext_type, canonical_name.to_string());
392 if self.handlers.contains_key(&key) {
393 return Err(RegistrationError::DuplicateName {
394 ext_type,
395 name: canonical_name.to_string(),
396 });
397 }
398
399 let type_url_key = (ext_type, type_url.clone());
401 if let Some(existing) = self.type_urls.get(&type_url_key)
402 && existing != canonical_name
403 {
404 return Err(RegistrationError::ConflictingTypeUrl {
405 type_url,
406 ext_type,
407 existing_name: existing.clone(),
408 });
409 }
410
411 self.handlers.insert(key, Arc::clone(&handler));
413 self.type_urls
414 .insert(type_url_key, canonical_name.to_string());
415 Ok(())
416 }
417
418 pub fn register_relation<T>(&mut self) -> Result<(), RegistrationError>
422 where
423 T: Extension,
424 {
425 self.register::<T>(ExtensionType::Relation)
426 }
427
428 pub fn register_enhancement<T>(&mut self) -> Result<(), RegistrationError>
435 where
436 T: Extension,
437 {
438 self.register::<T>(ExtensionType::Enhancement)
439 }
440
441 pub fn register_optimization<T>(&mut self) -> Result<(), RegistrationError>
448 where
449 T: Extension,
450 {
451 self.register::<T>(ExtensionType::Optimization)
452 }
453
454 pub fn parse_extension(
456 &self,
457 extension_name: &str,
458 args: &ExtensionArgs,
459 ) -> Result<Any, ExtensionError> {
460 self.parse_with_type(ExtensionType::Relation, extension_name, args)
461 }
462
463 pub fn parse_enhancement(
468 &self,
469 enhancement_name: &str,
470 args: &ExtensionArgs,
471 ) -> Result<Any, ExtensionError> {
472 self.parse_with_type(ExtensionType::Enhancement, enhancement_name, args)
473 }
474
475 pub fn parse_optimization(
480 &self,
481 optimization_name: &str,
482 args: &ExtensionArgs,
483 ) -> Result<Any, ExtensionError> {
484 self.parse_with_type(ExtensionType::Optimization, optimization_name, args)
485 }
486
487 fn parse_with_type(
489 &self,
490 ext_type: ExtensionType,
491 name: &str,
492 args: &ExtensionArgs,
493 ) -> Result<Any, ExtensionError> {
494 let key = (ext_type, name.to_string());
495 let handler = self
496 .handlers
497 .get(&key)
498 .ok_or_else(|| ExtensionError::NotFound {
499 name: name.to_string(),
500 })?;
501 handler.parse_detail(args)
502 }
503
504 pub fn decode(&self, detail: AnyRef<'_>) -> Result<(String, ExtensionArgs), ExtensionError> {
508 self.decode_with_type(ExtensionType::Relation, detail)
509 }
510
511 pub fn decode_enhancement(
519 &self,
520 detail: AnyRef<'_>,
521 ) -> Result<(String, ExtensionArgs), ExtensionError> {
522 self.decode_with_type(ExtensionType::Enhancement, detail)
523 }
524
525 pub fn decode_optimization(
533 &self,
534 detail: AnyRef<'_>,
535 ) -> Result<(String, ExtensionArgs), ExtensionError> {
536 self.decode_with_type(ExtensionType::Optimization, detail)
537 }
538
539 fn decode_with_type(
541 &self,
542 ext_type: ExtensionType,
543 detail: AnyRef<'_>,
544 ) -> Result<(String, ExtensionArgs), ExtensionError> {
545 let type_url_key = (ext_type, detail.type_url.to_string());
547 let extension_name =
548 self.type_urls
549 .get(&type_url_key)
550 .ok_or_else(|| ExtensionError::NotFound {
551 name: detail.type_url.to_string(),
552 })?;
553
554 let name_key = (ext_type, extension_name.clone());
556 let handler = self
557 .handlers
558 .get(&name_key)
559 .ok_or_else(|| ExtensionError::NotFound {
560 name: extension_name.clone(),
561 })?;
562
563 let args = handler.textify_detail(detail)?;
564
565 Ok((extension_name.clone(), args))
566 }
567
568 pub fn extension_names(&self, ext_type: ExtensionType) -> Vec<&str> {
570 let mut names: Vec<&str> = self
571 .type_urls
572 .iter()
573 .filter_map(|((t, _), name)| {
574 if *t == ext_type {
575 Some(name.as_str())
576 } else {
577 None
578 }
579 })
580 .collect();
581 names.sort_unstable();
582 names.dedup();
583 names
584 }
585
586 pub fn has_extension(&self, ext_type: ExtensionType, name: &str) -> bool {
588 self.handlers.contains_key(&(ext_type, name.to_string()))
589 }
590}
591
592impl fmt::Debug for ExtensionRegistry {
593 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
594 let mut keys: Vec<_> = self
595 .handlers
596 .keys()
597 .map(|(t, n)| (format!("{t:?}"), n.as_str()))
598 .collect();
599 keys.sort();
600 f.debug_struct("ExtensionRegistry")
601 .field("handlers", &keys)
602 .finish()
603 }
604}
605
606#[cfg(test)]
607mod tests {
608 use super::*;
609 use crate::extensions::ExtensionColumn;
610
611 struct TestExtension {
613 path: String,
614 batch_size: i64,
615 }
616
617 impl AnyConvertible for TestExtension {
619 fn to_any(&self) -> Result<Any, ExtensionError> {
620 let json_str = format!(
622 r#"{{"path":"{}","batch_size":{}}}"#,
623 self.path, self.batch_size
624 );
625 Ok(Any::new(Self::type_url(), json_str.into_bytes()))
626 }
627
628 fn type_url() -> String {
629 "test.TestExtension".to_string()
630 }
631
632 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
633 let json_str = String::from_utf8(any.value.to_vec())
635 .map_err(|e| ExtensionError::Custom(format!("Invalid UTF-8: {e}")))?;
636
637 if json_str.contains("path") && json_str.contains("batch_size") {
639 Ok(TestExtension {
640 path: "test.parquet".to_string(),
641 batch_size: 1024,
642 })
643 } else {
644 Err(ExtensionError::Custom("Missing fields".to_string()))
645 }
646 }
647 }
648
649 impl Explainable for TestExtension {
650 fn name() -> &'static str {
651 "TestExtension"
652 }
653
654 fn from_args(args: &ExtensionArgs) -> Result<Self, ExtensionError> {
655 let mut extractor = args.extractor();
656 let path: String = extractor.expect_named_arg::<&str>("path")?.to_string();
657 let batch_size: i64 = extractor.expect_named_arg("batch_size")?;
658 extractor.check_exhausted()?;
659
660 Ok(TestExtension {
661 path: path.to_string(),
662 batch_size,
663 })
664 }
665
666 fn to_args(&self) -> Result<ExtensionArgs, ExtensionError> {
667 let mut args = ExtensionArgs::default();
668 args.insert("path", self.path.clone());
669 args.insert("batch_size", self.batch_size);
670 Ok(args)
671 }
672 }
673
674 #[test]
675 fn test_extension_registry_basic() {
676 let mut registry = ExtensionRegistry::new();
677
678 assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 0);
680 assert!(!registry.has_extension(ExtensionType::Relation, "TestExtension"));
681
682 registry.register_relation::<TestExtension>().unwrap();
684
685 assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 1);
687 assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
688
689 let mut args = ExtensionArgs::default();
691 args.insert("path", "data.parquet");
692 args.insert("batch_size", 2048_i64);
693
694 let any = registry.parse_extension("TestExtension", &args).unwrap();
695 assert_eq!(any.type_url, "test.TestExtension");
696
697 let any_ref = any.as_ref();
698 let result = registry.decode(any_ref).unwrap();
699 assert_eq!(result.0, "TestExtension");
700 assert_eq!(
701 <&str>::try_from(result.1.named.get("path").unwrap()).unwrap(),
702 "test.parquet"
703 );
704 }
705
706 #[test]
707 fn test_extension_args() {
708 let mut args = ExtensionArgs::default();
709
710 args.insert("path", "data/*.parquet");
712 args.insert("batch_size", 1024_i64);
713
714 args.push(crate::textify::expressions::Reference(0));
716
717 args.output_columns.push(ExtensionColumn::Named {
719 name: "col1".to_string(),
720 r#type: crate::fixtures::parse_type("i32"),
721 });
722
723 let mut extractor = args.extractor();
725
726 let path = extractor.get_named_arg("path").unwrap();
727 assert_eq!(<&str>::try_from(path).unwrap(), "data/*.parquet");
728
729 let batch_size = extractor.get_named_arg("batch_size").unwrap();
730 assert_eq!(i64::try_from(batch_size).unwrap(), 1024);
731
732 assert!(extractor.check_exhausted().is_ok());
734
735 assert_eq!(args.positional.len(), 1);
736 assert_eq!(args.output_columns.len(), 1);
737 }
738
739 #[test]
740 fn test_extension_error_cases() {
741 let registry = ExtensionRegistry::new();
742
743 let args = ExtensionArgs::default();
745 let result = registry.parse_extension("NonExistent", &args);
746 assert!(matches!(result, Err(ExtensionError::NotFound { .. })));
747
748 let args = ExtensionArgs::default();
750 let mut extractor = args.extractor();
751 let result = extractor.get_named_arg("missing");
752 assert!(result.is_none());
753 assert!(extractor.check_exhausted().is_ok());
754
755 let mut args = ExtensionArgs::default();
757 args.insert("test", 42_i64);
758 let mut extractor = args.extractor();
759 let result = extractor.get_named_arg("test");
760 assert_eq!(i64::try_from(result.unwrap()).unwrap(), 42);
761 assert!(extractor.check_exhausted().is_ok());
762 }
763
764 struct TestEnhancement {
766 hint: String,
767 }
768
769 impl AnyConvertible for TestEnhancement {
770 fn to_any(&self) -> Result<Any, ExtensionError> {
771 let json_str = format!(r#"{{"hint":"{}"}}"#, self.hint);
772 Ok(Any::new(Self::type_url(), json_str.into_bytes()))
773 }
774
775 fn type_url() -> String {
776 "test.TestExtension".to_string()
778 }
779
780 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
781 let json_str = String::from_utf8(any.value.to_vec())
782 .map_err(|e| ExtensionError::Custom(format!("Invalid UTF-8: {e}")))?;
783 if json_str.contains("hint") {
784 Ok(TestEnhancement {
785 hint: "test_hint".to_string(),
786 })
787 } else {
788 Err(ExtensionError::Custom("Missing hint field".to_string()))
789 }
790 }
791 }
792
793 impl Explainable for TestEnhancement {
794 fn name() -> &'static str {
795 "TestEnhancement"
796 }
797
798 fn from_args(args: &ExtensionArgs) -> Result<Self, ExtensionError> {
799 let mut extractor = args.extractor();
800 let hint: String = extractor.expect_named_arg::<&str>("hint")?.to_string();
801 extractor.check_exhausted()?;
802 Ok(TestEnhancement { hint })
803 }
804
805 fn to_args(&self) -> Result<ExtensionArgs, ExtensionError> {
806 let mut args = ExtensionArgs::default();
807 args.insert("hint", self.hint.clone());
808 Ok(args)
809 }
810 }
811
812 #[test]
813 fn test_namespace_separation() {
814 let mut registry = ExtensionRegistry::new();
815
816 registry.register_relation::<TestExtension>().unwrap();
818 registry.register_enhancement::<TestEnhancement>().unwrap();
819
820 assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
822 assert!(registry.has_extension(ExtensionType::Enhancement, "TestEnhancement"));
823 assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 1);
824 assert_eq!(
825 registry.extension_names(ExtensionType::Enhancement).len(),
826 1
827 );
828
829 let mut ext_args = ExtensionArgs::default();
831 ext_args.insert("path", "data.parquet");
832 ext_args.insert("batch_size", 2048_i64);
833
834 let ext_any = registry
835 .parse_extension("TestExtension", &ext_args)
836 .unwrap();
837 assert_eq!(ext_any.type_url, "test.TestExtension");
838
839 let mut enh_args = ExtensionArgs::default();
841 enh_args.insert("hint", "optimize");
842
843 let enh_any = registry
844 .parse_enhancement("TestEnhancement", &enh_args)
845 .unwrap();
846 assert_eq!(enh_any.type_url, "test.TestExtension"); let enh_ref = enh_any.as_ref();
850 let (name, args) = registry.decode_enhancement(enh_ref).unwrap();
851 assert_eq!(name, "TestEnhancement");
852 assert_eq!(
853 <&str>::try_from(args.named.get("hint").unwrap()).unwrap(),
854 "test_hint"
855 );
856 }
857
858 #[test]
859 fn test_enhancement_duplicate_registration_returns_error() {
860 let mut registry = ExtensionRegistry::new();
861 registry.register_enhancement::<TestEnhancement>().unwrap();
862 let result = registry.register_enhancement::<TestEnhancement>();
863 assert!(matches!(
864 result,
865 Err(RegistrationError::DuplicateName { .. })
866 ));
867 }
868
869 #[test]
870 fn test_enhancement_not_found_error() {
871 let registry = ExtensionRegistry::new();
872 let args = ExtensionArgs::default();
873 let result = registry.parse_enhancement("NonExistentEnhancement", &args);
874 assert!(matches!(result, Err(ExtensionError::NotFound { .. })));
875 }
876
877 struct ConflictingExtension;
880
881 impl AnyConvertible for ConflictingExtension {
882 fn to_any(&self) -> Result<Any, ExtensionError> {
883 Ok(Any::new(Self::type_url(), vec![]))
884 }
885
886 fn type_url() -> String {
887 "test.TestExtension".to_string()
889 }
890
891 fn from_any<'a>(_any: AnyRef<'a>) -> Result<Self, ExtensionError> {
892 Ok(ConflictingExtension)
893 }
894 }
895
896 impl Explainable for ConflictingExtension {
897 fn name() -> &'static str {
898 "ConflictingExtension"
899 }
900
901 fn from_args(_args: &ExtensionArgs) -> Result<Self, ExtensionError> {
902 Ok(ConflictingExtension)
903 }
904
905 fn to_args(&self) -> Result<ExtensionArgs, ExtensionError> {
906 Ok(ExtensionArgs::default())
907 }
908 }
909
910 #[test]
911 fn test_conflicting_type_url_leaves_registry_unchanged() {
912 let mut registry = ExtensionRegistry::new();
913 registry.register_relation::<TestExtension>().unwrap();
914
915 let result = registry.register_relation::<ConflictingExtension>();
917 assert!(matches!(
918 result,
919 Err(RegistrationError::ConflictingTypeUrl { .. })
920 ));
921
922 assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
924 assert!(!registry.has_extension(ExtensionType::Relation, "ConflictingExtension"));
925 assert_eq!(
926 registry.extension_names(ExtensionType::Relation),
927 vec!["TestExtension"]
928 );
929 }
930}