1use std::collections::HashMap;
86use std::fmt;
87use std::marker::PhantomData;
88use std::sync::Arc;
89
90use substrait::proto::NamedStruct;
91use substrait::proto::r#type::{Nullability, Struct};
92use thiserror::Error;
93
94use crate::extensions::any::{Any, AnyRef};
95use crate::extensions::args::{ArgsAccess, ExtensionArgs, ExtensionColumn, ExtensionValueKind};
96
97#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
99pub enum ExtensionType {
100 Relation,
102 ExtensionTable,
104 Enhancement,
106 Optimization,
108}
109
110#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112pub struct ExtensionInput {
113 emitted_column_count: usize,
114}
115
116impl ExtensionInput {
117 pub(crate) fn new(emitted_column_count: usize) -> Self {
118 Self {
119 emitted_column_count,
120 }
121 }
122
123 pub fn emitted_column_count(&self) -> usize {
125 self.emitted_column_count
126 }
127}
128
129#[derive(Debug, Clone, Copy, Default)]
135pub struct ExtensionContext<'a> {
136 inputs: &'a [ExtensionInput],
137}
138
139impl<'a> ExtensionContext<'a> {
140 pub(crate) fn new(inputs: &'a [ExtensionInput]) -> Self {
141 Self { inputs }
142 }
143
144 pub fn inputs(&self) -> &'a [ExtensionInput] {
146 self.inputs
147 }
148}
149
150#[derive(Debug, Error, Clone)]
152pub enum RegistrationError {
153 #[error("{ext_type:?} extension '{name}' already registered")]
154 DuplicateName {
155 ext_type: ExtensionType,
156 name: String,
157 },
158
159 #[error("Type URL '{type_url}' already registered to {ext_type:?} extension '{existing_name}'")]
160 ConflictingTypeUrl {
161 type_url: String,
162 ext_type: ExtensionType,
163 existing_name: String,
164 },
165}
166
167#[derive(Debug, Error, Clone)]
169pub enum ExtensionError {
170 #[error("Extension '{name}' not found in registry")]
172 NotFound { name: String },
173
174 #[error("Missing required argument: {name}")]
176 MissingArgument { name: String },
177
178 #[error("Invalid named argument '{name}': {source}")]
180 NamedArgumentConversion {
181 name: String,
182 #[source]
183 source: Box<ExtensionError>,
184 },
185
186 #[error("Cannot convert argument to {expected}: {source}")]
188 ArgumentConversion {
189 expected: ExtensionValueKind,
190 #[source]
191 source: Box<ExtensionError>,
192 },
193
194 #[error("Invalid argument: expected {expected}, got {actual}")]
196 InvalidArgumentType {
197 expected: ExtensionValueKind,
198 actual: ExtensionValueKind,
199 },
200
201 #[error("Invalid argument: {0}")]
207 InvalidArgument(String),
208
209 #[error("Type URL mismatch: expected {expected}, got {actual}")]
211 TypeUrlMismatch { expected: String, actual: String },
212
213 #[error("Failed to decode protobuf message")]
215 DecodeFailed(#[source] prost::DecodeError),
216
217 #[error("Failed to encode protobuf message")]
219 EncodeFailed(#[source] prost::EncodeError),
220
221 #[error("Extension detail is missing")]
223 MissingDetail,
224
225 #[error("{0}")]
227 Custom(String),
228}
229
230pub trait AnyConvertible: Sized {
234 fn to_any(&self) -> Result<Any, ExtensionError>;
236
237 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError>;
239
240 fn type_url() -> String;
244}
245
246impl<T> AnyConvertible for T
248where
249 T: prost::Message + prost::Name + Default,
250{
251 fn to_any(&self) -> Result<Any, ExtensionError> {
252 Any::encode(self)
253 }
254
255 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
256 any.decode()
257 }
258
259 fn type_url() -> String {
260 T::type_url()
261 }
262}
263
264pub trait ExtensionProtoConvert<T> {
275 fn convert(&self) -> Result<T, ExtensionError>;
277}
278
279impl ExtensionProtoConvert<NamedStruct> for [ExtensionColumn] {
280 fn convert(&self) -> Result<NamedStruct, ExtensionError> {
281 let mut names = Vec::with_capacity(self.len());
282 let mut types = Vec::with_capacity(self.len());
283 for col in self {
284 match col {
285 ExtensionColumn::Named { name, r#type: ty } => {
286 names.push(name.clone());
287 types.push(ty.clone());
288 }
289 other => {
290 return Err(ExtensionError::InvalidArgument(format!(
291 "Expected named column, got {other:?}"
292 )));
293 }
294 }
295 }
296 Ok(NamedStruct {
297 names,
298 r#struct: Some(Struct {
299 types,
300 type_variation_reference: 0,
301 nullability: Nullability::Required as i32,
305 }),
306 })
307 }
308}
309
310impl ExtensionProtoConvert<Vec<ExtensionColumn>> for NamedStruct {
311 fn convert(&self) -> Result<Vec<ExtensionColumn>, ExtensionError> {
312 let types = self
313 .r#struct
314 .as_ref()
315 .map(|s| s.types.as_slice())
316 .unwrap_or_default();
317 if self.names.len() != types.len() {
318 return Err(ExtensionError::InvalidArgument(format!(
319 "NamedStruct has {} names but {} types",
320 self.names.len(),
321 types.len()
322 )));
323 }
324 Ok(self
325 .names
326 .iter()
327 .zip(types.iter())
328 .map(|(name, ty)| ExtensionColumn::Named {
329 name: name.clone(),
330 r#type: ty.clone(),
331 })
332 .collect())
333 }
334}
335
336pub trait Explainable: Sized {
341 fn name() -> &'static str;
343
344 fn from_args(args: &mut ArgsAccess<'_>) -> Result<Self, ExtensionError>;
352
353 fn to_args(&self, context: &ExtensionContext<'_>) -> Result<ExtensionArgs, ExtensionError>;
357}
358
359trait ExtensionConverter: Send + Sync {
373 fn parse_detail(&self, args: &ExtensionArgs) -> Result<Any, ExtensionError>;
374
375 fn textify_detail(
376 &self,
377 detail: AnyRef<'_>,
378 context: &ExtensionContext<'_>,
379 ) -> Result<ExtensionArgs, ExtensionError>;
380}
381
382struct ExtensionAdapter<T>(PhantomData<T>);
399
400impl<T: AnyConvertible + Explainable + Send + Sync> ExtensionConverter for ExtensionAdapter<T> {
401 fn parse_detail(&self, args: &ExtensionArgs) -> Result<Any, ExtensionError> {
402 args.parse::<T>()?.to_any()
403 }
404
405 fn textify_detail(
406 &self,
407 detail: AnyRef<'_>,
408 context: &ExtensionContext<'_>,
409 ) -> Result<ExtensionArgs, ExtensionError> {
410 let owned_any = Any::new(detail.type_url.to_string(), detail.value.to_vec());
411 T::from_any(owned_any.as_ref())?.to_args(context)
412 }
413}
414
415pub trait Extension: AnyConvertible + Explainable + Send + Sync + 'static {}
416
417impl<T> Extension for T where T: AnyConvertible + Explainable + Send + Sync + 'static {}
418
419#[derive(Default, Clone)]
421pub struct ExtensionRegistry {
422 handlers: HashMap<(ExtensionType, String), Arc<dyn ExtensionConverter>>,
424 type_urls: HashMap<(ExtensionType, String), String>,
426 descriptors: Vec<Vec<u8>>,
432}
433
434impl ExtensionRegistry {
435 pub fn new() -> Self {
437 Self {
438 handlers: HashMap::new(),
439 type_urls: HashMap::new(),
440 descriptors: Vec::new(),
441 }
442 }
443
444 pub fn add_descriptor(&mut self, bytes: Vec<u8>) {
452 self.descriptors.push(bytes);
453 }
454
455 pub fn descriptors(&self) -> Vec<&[u8]> {
457 self.descriptors.iter().map(|b| b.as_slice()).collect()
458 }
459
460 fn register<T>(&mut self, ext_type: ExtensionType) -> Result<(), RegistrationError>
462 where
463 T: Extension,
464 {
465 let canonical_name = T::name();
466 let type_url = T::type_url();
467 let handler: Arc<dyn ExtensionConverter> = Arc::new(ExtensionAdapter::<T>(PhantomData));
468
469 let key = (ext_type, canonical_name.to_string());
470 if self.handlers.contains_key(&key) {
471 return Err(RegistrationError::DuplicateName {
472 ext_type,
473 name: canonical_name.to_string(),
474 });
475 }
476
477 let type_url_key = (ext_type, type_url.clone());
479 if let Some(existing) = self.type_urls.get(&type_url_key)
480 && existing != canonical_name
481 {
482 return Err(RegistrationError::ConflictingTypeUrl {
483 type_url,
484 ext_type,
485 existing_name: existing.clone(),
486 });
487 }
488
489 self.handlers.insert(key, Arc::clone(&handler));
491 self.type_urls
492 .insert(type_url_key, canonical_name.to_string());
493 Ok(())
494 }
495
496 pub fn register_relation<T>(&mut self) -> Result<(), RegistrationError>
500 where
501 T: Extension,
502 {
503 self.register::<T>(ExtensionType::Relation)
504 }
505
506 pub fn register_extension_table<T>(&mut self) -> Result<(), RegistrationError>
514 where
515 T: Extension,
516 {
517 self.register::<T>(ExtensionType::ExtensionTable)
518 }
519
520 pub fn register_enhancement<T>(&mut self) -> Result<(), RegistrationError>
527 where
528 T: Extension,
529 {
530 self.register::<T>(ExtensionType::Enhancement)
531 }
532
533 pub fn register_optimization<T>(&mut self) -> Result<(), RegistrationError>
540 where
541 T: Extension,
542 {
543 self.register::<T>(ExtensionType::Optimization)
544 }
545
546 pub fn parse_extension(
548 &self,
549 extension_name: &str,
550 args: &ExtensionArgs,
551 ) -> Result<Any, ExtensionError> {
552 self.parse_with_type(ExtensionType::Relation, extension_name, args)
553 }
554
555 pub fn parse_extension_table(
560 &self,
561 extension_table_name: &str,
562 args: &ExtensionArgs,
563 ) -> Result<Any, ExtensionError> {
564 self.parse_with_type(ExtensionType::ExtensionTable, extension_table_name, args)
565 }
566
567 pub fn parse_enhancement(
572 &self,
573 enhancement_name: &str,
574 args: &ExtensionArgs,
575 ) -> Result<Any, ExtensionError> {
576 self.parse_with_type(ExtensionType::Enhancement, enhancement_name, args)
577 }
578
579 pub fn parse_optimization(
584 &self,
585 optimization_name: &str,
586 args: &ExtensionArgs,
587 ) -> Result<Any, ExtensionError> {
588 self.parse_with_type(ExtensionType::Optimization, optimization_name, args)
589 }
590
591 fn parse_with_type(
593 &self,
594 ext_type: ExtensionType,
595 name: &str,
596 args: &ExtensionArgs,
597 ) -> Result<Any, ExtensionError> {
598 let key = (ext_type, name.to_string());
599 let handler = self
600 .handlers
601 .get(&key)
602 .ok_or_else(|| ExtensionError::NotFound {
603 name: name.to_string(),
604 })?;
605 handler.parse_detail(args)
606 }
607
608 pub fn decode(&self, detail: AnyRef<'_>) -> Result<(String, ExtensionArgs), ExtensionError> {
612 self.decode_with_type(ExtensionType::Relation, detail)
613 }
614
615 pub(crate) fn decode_with_context(
617 &self,
618 detail: AnyRef<'_>,
619 context: &ExtensionContext<'_>,
620 ) -> Result<(String, ExtensionArgs), ExtensionError> {
621 self.decode_with_type_and_context(ExtensionType::Relation, detail, context)
622 }
623
624 pub fn decode_extension_table(
630 &self,
631 detail: AnyRef<'_>,
632 ) -> Result<(String, ExtensionArgs), ExtensionError> {
633 self.decode_with_type(ExtensionType::ExtensionTable, detail)
634 }
635
636 pub fn decode_enhancement(
644 &self,
645 detail: AnyRef<'_>,
646 ) -> Result<(String, ExtensionArgs), ExtensionError> {
647 self.decode_with_type(ExtensionType::Enhancement, detail)
648 }
649
650 pub fn decode_optimization(
658 &self,
659 detail: AnyRef<'_>,
660 ) -> Result<(String, ExtensionArgs), ExtensionError> {
661 self.decode_with_type(ExtensionType::Optimization, detail)
662 }
663
664 fn decode_with_type(
666 &self,
667 ext_type: ExtensionType,
668 detail: AnyRef<'_>,
669 ) -> Result<(String, ExtensionArgs), ExtensionError> {
670 self.decode_with_type_and_context(ext_type, detail, &ExtensionContext::default())
671 }
672
673 fn decode_with_type_and_context(
674 &self,
675 ext_type: ExtensionType,
676 detail: AnyRef<'_>,
677 context: &ExtensionContext<'_>,
678 ) -> Result<(String, ExtensionArgs), ExtensionError> {
679 let type_url_key = (ext_type, detail.type_url.to_string());
681 let extension_name =
682 self.type_urls
683 .get(&type_url_key)
684 .ok_or_else(|| ExtensionError::NotFound {
685 name: detail.type_url.to_string(),
686 })?;
687
688 let name_key = (ext_type, extension_name.clone());
690 let handler = self
691 .handlers
692 .get(&name_key)
693 .ok_or_else(|| ExtensionError::NotFound {
694 name: extension_name.clone(),
695 })?;
696
697 let args = handler.textify_detail(detail, context)?;
698
699 Ok((extension_name.clone(), args))
700 }
701
702 pub fn extension_names(&self, ext_type: ExtensionType) -> Vec<&str> {
704 let mut names: Vec<&str> = self
705 .type_urls
706 .iter()
707 .filter_map(|((t, _), name)| {
708 if *t == ext_type {
709 Some(name.as_str())
710 } else {
711 None
712 }
713 })
714 .collect();
715 names.sort_unstable();
716 names.dedup();
717 names
718 }
719
720 pub fn has_extension(&self, ext_type: ExtensionType, name: &str) -> bool {
722 self.handlers.contains_key(&(ext_type, name.to_string()))
723 }
724}
725
726impl fmt::Debug for ExtensionRegistry {
727 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
728 let mut keys: Vec<_> = self
729 .handlers
730 .keys()
731 .map(|(t, n)| (format!("{t:?}"), n.as_str()))
732 .collect();
733 keys.sort();
734 f.debug_struct("ExtensionRegistry")
735 .field("handlers", &keys)
736 .finish()
737 }
738}
739
740#[cfg(test)]
741mod tests {
742 use super::*;
743 use crate::extensions::ExtensionColumn;
744 use crate::fixtures::parse_type;
745 use crate::textify::expressions::Reference;
746
747 struct TestExtension {
749 path: String,
750 batch_size: i64,
751 }
752
753 impl AnyConvertible for TestExtension {
755 fn to_any(&self) -> Result<Any, ExtensionError> {
756 let json_str = format!(
758 r#"{{"path":"{}","batch_size":{}}}"#,
759 self.path, self.batch_size
760 );
761 Ok(Any::new(Self::type_url(), json_str.into_bytes()))
762 }
763
764 fn type_url() -> String {
765 "test.TestExtension".to_string()
766 }
767
768 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
769 let json_str = String::from_utf8(any.value.to_vec())
771 .map_err(|e| ExtensionError::Custom(format!("Invalid UTF-8: {e}")))?;
772
773 if json_str.contains("path") && json_str.contains("batch_size") {
775 Ok(TestExtension {
776 path: "test.parquet".to_string(),
777 batch_size: 1024,
778 })
779 } else {
780 Err(ExtensionError::Custom("Missing fields".to_string()))
781 }
782 }
783 }
784
785 impl Explainable for TestExtension {
786 fn name() -> &'static str {
787 "TestExtension"
788 }
789
790 fn from_args(args: &mut ArgsAccess<'_>) -> Result<Self, ExtensionError> {
791 let path: String = args.expect_named::<&str>("path")?.to_string();
792 let batch_size: i64 = args.expect_named("batch_size")?;
793
794 Ok(TestExtension {
795 path: path.to_string(),
796 batch_size,
797 })
798 }
799
800 fn to_args(&self, context: &ExtensionContext<'_>) -> Result<ExtensionArgs, ExtensionError> {
801 let mut args = ExtensionArgs::default();
802 args.insert("path", self.path.clone());
803 args.insert("batch_size", self.batch_size);
804 if !context.inputs().is_empty() {
805 args.insert("input_count", context.inputs().len() as i64);
806 }
807 Ok(args)
808 }
809 }
810
811 #[test]
812 fn test_extension_registry_basic() {
813 let mut registry = ExtensionRegistry::new();
814
815 assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 0);
817 assert_eq!(
818 registry
819 .extension_names(ExtensionType::ExtensionTable)
820 .len(),
821 0
822 );
823 assert!(!registry.has_extension(ExtensionType::Relation, "TestExtension"));
824
825 registry.register_relation::<TestExtension>().unwrap();
827
828 assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 1);
830 assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
831
832 let mut args = ExtensionArgs::default();
834 args.insert("path", "data.parquet");
835 args.insert("batch_size", 2048_i64);
836
837 let any = registry.parse_extension("TestExtension", &args).unwrap();
838 assert_eq!(any.type_url, "test.TestExtension");
839
840 let any_ref = any.as_ref();
841 let result = registry.decode(any_ref).unwrap();
842 assert_eq!(result.0, "TestExtension");
843 assert_eq!(
844 <&str>::try_from(result.1.named.get("path").unwrap()).unwrap(),
845 "test.parquet"
846 );
847 assert!(!result.1.named.contains_key("input_count"));
848 }
849
850 #[test]
851 fn test_extension_registry_decode_with_context() {
852 let mut registry = ExtensionRegistry::new();
853 registry.register_relation::<TestExtension>().unwrap();
854
855 let mut args = ExtensionArgs::default();
856 args.insert("path", "data.parquet");
857 args.insert("batch_size", 2048_i64);
858 let any = registry.parse_extension("TestExtension", &args).unwrap();
859 let inputs = [ExtensionInput::new(4)];
860 let context = ExtensionContext::new(&inputs);
861
862 let (_, decoded_args) = registry
863 .decode_with_context(any.as_ref(), &context)
864 .unwrap();
865 assert_eq!(
866 i64::try_from(decoded_args.named.get("input_count").unwrap()).unwrap(),
867 1
868 );
869 }
870
871 #[test]
872 fn argument_error_takes_precedence_over_unhandled_arguments() {
873 let mut registry = ExtensionRegistry::new();
874 registry.register_relation::<TestExtension>().unwrap();
875
876 let mut args = ExtensionArgs::default();
877 args.insert("unexpected", true);
878
879 let error = registry
880 .parse_extension("TestExtension", &args)
881 .expect_err("missing path should return an error");
882
883 assert!(matches!(
884 error,
885 ExtensionError::MissingArgument { name } if name == "path"
886 ));
887 }
888
889 #[test]
890 fn successful_decode_rejects_unhandled_named_arguments() {
891 let mut registry = ExtensionRegistry::new();
892 registry.register_relation::<TestExtension>().unwrap();
893
894 let mut args = ExtensionArgs::default();
895 args.insert("path", "data.parquet");
896 args.insert("batch_size", 2048_i64);
897 args.insert("unexpected", true);
898
899 let error = registry
900 .parse_extension("TestExtension", &args)
901 .expect_err("unhandled argument should fail");
902
903 assert_eq!(
904 error.to_string(),
905 "Invalid argument: Unknown named arguments: unexpected"
906 );
907 }
908
909 #[test]
910 fn successful_decode_rejects_unhandled_positional_arguments() {
911 let mut registry = ExtensionRegistry::new();
912 registry.register_relation::<TestExtension>().unwrap();
913
914 let mut args = ExtensionArgs::default();
915 args.push(true);
916 args.insert("path", "data.parquet");
917 args.insert("batch_size", 2048_i64);
918
919 let error = registry
920 .parse_extension("TestExtension", &args)
921 .expect_err("unhandled positional argument should fail");
922
923 assert_eq!(
924 error.to_string(),
925 "Invalid argument: Unhandled positional arguments: 1"
926 );
927 }
928
929 #[test]
930 fn test_extension_table_registry_basic() {
931 let mut registry = ExtensionRegistry::new();
932
933 registry
934 .register_extension_table::<TestExtension>()
935 .unwrap();
936
937 assert_eq!(
938 registry.extension_names(ExtensionType::ExtensionTable),
939 vec!["TestExtension"]
940 );
941 assert!(registry.has_extension(ExtensionType::ExtensionTable, "TestExtension"));
942
943 let mut args = ExtensionArgs::default();
944 args.insert("path", "data.parquet");
945 args.insert("batch_size", 2048_i64);
946
947 let any = registry
948 .parse_extension_table("TestExtension", &args)
949 .unwrap();
950 assert_eq!(any.type_url, "test.TestExtension");
951
952 let (name, decoded_args) = registry.decode_extension_table(any.as_ref()).unwrap();
953 assert_eq!(name, "TestExtension");
954 assert_eq!(
955 <&str>::try_from(decoded_args.named.get("path").unwrap()).unwrap(),
956 "test.parquet"
957 );
958 assert!(!decoded_args.named.contains_key("input_count"));
959 }
960
961 #[test]
962 fn test_extension_args() {
963 let mut args = ExtensionArgs::default();
964
965 args.insert("path", "data/*.parquet");
967 args.insert("batch_size", 1024_i64);
968
969 args.push(Reference(0));
971
972 args.output_columns.push(ExtensionColumn::Named {
974 name: "col1".to_string(),
975 r#type: parse_type("i32"),
976 });
977
978 let mut access = ArgsAccess::new(&args);
979
980 let path = access.get_named_arg("path").unwrap();
981 assert_eq!(<&str>::try_from(path).unwrap(), "data/*.parquet");
982
983 let batch_size = access.get_named_arg("batch_size").unwrap();
984 assert_eq!(i64::try_from(batch_size).unwrap(), 1024);
985 assert_eq!(access.positional().len(), 1);
986
987 assert!(access.finish().is_ok());
989
990 assert_eq!(args.positional.len(), 1);
991 assert_eq!(args.output_columns.len(), 1);
992 }
993
994 #[test]
995 fn test_extension_error_cases() {
996 let registry = ExtensionRegistry::new();
997
998 let args = ExtensionArgs::default();
1000 let result = registry.parse_extension("NonExistent", &args);
1001 assert!(matches!(result, Err(ExtensionError::NotFound { .. })));
1002
1003 let args = ExtensionArgs::default();
1004 let mut access = ArgsAccess::new(&args);
1005 let result = access.get_named_arg("missing");
1006 assert!(result.is_none());
1007 assert!(access.finish().is_ok());
1008
1009 let mut args = ExtensionArgs::default();
1010 args.insert("test", 42_i64);
1011 let mut access = ArgsAccess::new(&args);
1012 let result = access.get_named_arg("test");
1013 assert_eq!(i64::try_from(result.unwrap()).unwrap(), 42);
1014 assert!(access.finish().is_ok());
1015 }
1016
1017 struct TestEnhancement {
1019 hint: String,
1020 }
1021
1022 impl AnyConvertible for TestEnhancement {
1023 fn to_any(&self) -> Result<Any, ExtensionError> {
1024 let json_str = format!(r#"{{"hint":"{}"}}"#, self.hint);
1025 Ok(Any::new(Self::type_url(), json_str.into_bytes()))
1026 }
1027
1028 fn type_url() -> String {
1029 "test.TestExtension".to_string()
1031 }
1032
1033 fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
1034 let json_str = String::from_utf8(any.value.to_vec())
1035 .map_err(|e| ExtensionError::Custom(format!("Invalid UTF-8: {e}")))?;
1036 if json_str.contains("hint") {
1037 Ok(TestEnhancement {
1038 hint: "test_hint".to_string(),
1039 })
1040 } else {
1041 Err(ExtensionError::Custom("Missing hint field".to_string()))
1042 }
1043 }
1044 }
1045
1046 impl Explainable for TestEnhancement {
1047 fn name() -> &'static str {
1048 "TestEnhancement"
1049 }
1050
1051 fn from_args(args: &mut ArgsAccess<'_>) -> Result<Self, ExtensionError> {
1052 let hint: String = args.expect_named::<&str>("hint")?.to_string();
1053 Ok(TestEnhancement { hint })
1054 }
1055
1056 fn to_args(
1057 &self,
1058 _context: &ExtensionContext<'_>,
1059 ) -> Result<ExtensionArgs, ExtensionError> {
1060 let mut args = ExtensionArgs::default();
1061 args.insert("hint", self.hint.clone());
1062 Ok(args)
1063 }
1064 }
1065
1066 #[test]
1067 fn test_namespace_separation() {
1068 let mut registry = ExtensionRegistry::new();
1069
1070 registry.register_relation::<TestExtension>().unwrap();
1072 registry
1073 .register_extension_table::<TestExtension>()
1074 .unwrap();
1075 registry.register_enhancement::<TestEnhancement>().unwrap();
1076
1077 assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
1079 assert!(registry.has_extension(ExtensionType::ExtensionTable, "TestExtension"));
1080 assert!(registry.has_extension(ExtensionType::Enhancement, "TestEnhancement"));
1081 assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 1);
1082 assert_eq!(
1083 registry
1084 .extension_names(ExtensionType::ExtensionTable)
1085 .len(),
1086 1
1087 );
1088 assert_eq!(
1089 registry.extension_names(ExtensionType::Enhancement).len(),
1090 1
1091 );
1092
1093 let mut ext_args = ExtensionArgs::default();
1095 ext_args.insert("path", "data.parquet");
1096 ext_args.insert("batch_size", 2048_i64);
1097
1098 let ext_any = registry
1099 .parse_extension("TestExtension", &ext_args)
1100 .unwrap();
1101 assert_eq!(ext_any.type_url, "test.TestExtension");
1102
1103 let table_any = registry
1105 .parse_extension_table("TestExtension", &ext_args)
1106 .unwrap();
1107 assert_eq!(table_any.type_url, "test.TestExtension");
1108
1109 let mut enh_args = ExtensionArgs::default();
1111 enh_args.insert("hint", "optimize");
1112
1113 let enh_any = registry
1114 .parse_enhancement("TestEnhancement", &enh_args)
1115 .unwrap();
1116 assert_eq!(enh_any.type_url, "test.TestExtension"); let enh_ref = enh_any.as_ref();
1120 let (name, args) = registry.decode_enhancement(enh_ref).unwrap();
1121 assert_eq!(name, "TestEnhancement");
1122 assert_eq!(
1123 <&str>::try_from(args.named.get("hint").unwrap()).unwrap(),
1124 "test_hint"
1125 );
1126 }
1127
1128 #[test]
1129 fn test_enhancement_duplicate_registration_returns_error() {
1130 let mut registry = ExtensionRegistry::new();
1131 registry.register_enhancement::<TestEnhancement>().unwrap();
1132 let result = registry.register_enhancement::<TestEnhancement>();
1133 assert!(matches!(
1134 result,
1135 Err(RegistrationError::DuplicateName { .. })
1136 ));
1137 }
1138
1139 #[test]
1140 fn test_extension_table_duplicate_registration_returns_error() {
1141 let mut registry = ExtensionRegistry::new();
1142 registry
1143 .register_extension_table::<TestExtension>()
1144 .unwrap();
1145 let result = registry.register_extension_table::<TestExtension>();
1146 assert!(matches!(
1147 result,
1148 Err(RegistrationError::DuplicateName { .. })
1149 ));
1150 }
1151
1152 #[test]
1153 fn test_extension_table_not_found_error() {
1154 let registry = ExtensionRegistry::new();
1155 let args = ExtensionArgs::default();
1156 let result = registry.parse_extension_table("NonExistentExtensionTable", &args);
1157 assert!(matches!(result, Err(ExtensionError::NotFound { .. })));
1158 }
1159
1160 #[test]
1161 fn test_enhancement_not_found_error() {
1162 let registry = ExtensionRegistry::new();
1163 let args = ExtensionArgs::default();
1164 let result = registry.parse_enhancement("NonExistentEnhancement", &args);
1165 assert!(matches!(result, Err(ExtensionError::NotFound { .. })));
1166 }
1167
1168 struct ConflictingExtension;
1171
1172 impl AnyConvertible for ConflictingExtension {
1173 fn to_any(&self) -> Result<Any, ExtensionError> {
1174 Ok(Any::new(Self::type_url(), vec![]))
1175 }
1176
1177 fn type_url() -> String {
1178 "test.TestExtension".to_string()
1180 }
1181
1182 fn from_any<'a>(_any: AnyRef<'a>) -> Result<Self, ExtensionError> {
1183 Ok(ConflictingExtension)
1184 }
1185 }
1186
1187 impl Explainable for ConflictingExtension {
1188 fn name() -> &'static str {
1189 "ConflictingExtension"
1190 }
1191
1192 fn from_args(_args: &mut ArgsAccess<'_>) -> Result<Self, ExtensionError> {
1193 Ok(ConflictingExtension)
1194 }
1195
1196 fn to_args(
1197 &self,
1198 _context: &ExtensionContext<'_>,
1199 ) -> Result<ExtensionArgs, ExtensionError> {
1200 Ok(ExtensionArgs::default())
1201 }
1202 }
1203
1204 #[test]
1205 fn test_conflicting_type_url_leaves_registry_unchanged() {
1206 let mut registry = ExtensionRegistry::new();
1207 registry.register_relation::<TestExtension>().unwrap();
1208
1209 let result = registry.register_relation::<ConflictingExtension>();
1211 assert!(matches!(
1212 result,
1213 Err(RegistrationError::ConflictingTypeUrl { .. })
1214 ));
1215
1216 assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
1218 assert!(!registry.has_extension(ExtensionType::Relation, "ConflictingExtension"));
1219 assert_eq!(
1220 registry.extension_names(ExtensionType::Relation),
1221 vec!["TestExtension"]
1222 );
1223 }
1224
1225 #[test]
1226 fn test_extension_table_conflicting_type_url_leaves_registry_unchanged() {
1227 let mut registry = ExtensionRegistry::new();
1228 registry
1229 .register_extension_table::<TestExtension>()
1230 .unwrap();
1231
1232 let result = registry.register_extension_table::<ConflictingExtension>();
1234 assert!(matches!(
1235 result,
1236 Err(RegistrationError::ConflictingTypeUrl { .. })
1237 ));
1238
1239 assert!(registry.has_extension(ExtensionType::ExtensionTable, "TestExtension"));
1241 assert!(!registry.has_extension(ExtensionType::ExtensionTable, "ConflictingExtension"));
1242 assert_eq!(
1243 registry.extension_names(ExtensionType::ExtensionTable),
1244 vec!["TestExtension"]
1245 );
1246 }
1247}