use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;
use thiserror::Error;
use crate::extensions::any::{Any, AnyRef};
use crate::extensions::args::ExtensionArgs;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ExtensionType {
Relation,
Enhancement,
Optimization,
}
#[derive(Debug, Error, Clone)]
pub enum RegistrationError {
#[error("{ext_type:?} extension '{name}' already registered")]
DuplicateName {
ext_type: ExtensionType,
name: String,
},
#[error("Type URL '{type_url}' already registered to {ext_type:?} extension '{existing_name}'")]
ConflictingTypeUrl {
type_url: String,
ext_type: ExtensionType,
existing_name: String,
},
}
#[derive(Debug, Error, Clone)]
pub enum ExtensionError {
#[error("Extension '{name}' not found in registry")]
NotFound { name: String },
#[error("Missing required argument: {name}")]
MissingArgument { name: String },
#[error("Invalid argument: {0}")]
InvalidArgument(String),
#[error("Type URL mismatch: expected {expected}, got {actual}")]
TypeUrlMismatch { expected: String, actual: String },
#[error("Failed to decode protobuf message")]
DecodeFailed(#[source] prost::DecodeError),
#[error("Failed to encode protobuf message")]
EncodeFailed(#[source] prost::EncodeError),
#[error("Extension detail is missing")]
MissingDetail,
#[error("{0}")]
Custom(String),
}
pub trait AnyConvertible: Sized {
fn to_any(&self) -> Result<Any, ExtensionError>;
fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError>;
fn type_url() -> String;
}
impl<T> AnyConvertible for T
where
T: prost::Message + prost::Name + Default,
{
fn to_any(&self) -> Result<Any, ExtensionError> {
Any::encode(self)
}
fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
any.decode()
}
fn type_url() -> String {
T::type_url()
}
}
pub trait Explainable: Sized {
fn name() -> &'static str;
fn from_args(args: &ExtensionArgs) -> Result<Self, ExtensionError>;
fn to_args(&self) -> Result<ExtensionArgs, ExtensionError>;
}
trait ExtensionConverter: Send + Sync {
fn parse_detail(&self, args: &ExtensionArgs) -> Result<Any, ExtensionError>;
fn textify_detail(&self, detail: AnyRef<'_>) -> Result<ExtensionArgs, ExtensionError>;
}
struct ExtensionAdapter<T>(std::marker::PhantomData<T>);
impl<T: AnyConvertible + Explainable + Send + Sync> ExtensionConverter for ExtensionAdapter<T> {
fn parse_detail(&self, args: &ExtensionArgs) -> Result<Any, ExtensionError> {
T::from_args(args)?.to_any()
}
fn textify_detail(&self, detail: AnyRef<'_>) -> Result<ExtensionArgs, ExtensionError> {
let owned_any = Any::new(detail.type_url.to_string(), detail.value.to_vec());
T::from_any(owned_any.as_ref())?.to_args()
}
}
pub trait Extension: AnyConvertible + Explainable + Send + Sync + 'static {}
impl<T> Extension for T where T: AnyConvertible + Explainable + Send + Sync + 'static {}
#[derive(Default, Clone)]
pub struct ExtensionRegistry {
handlers: HashMap<(ExtensionType, String), Arc<dyn ExtensionConverter>>,
type_urls: HashMap<(ExtensionType, String), String>,
descriptors: Vec<Vec<u8>>,
}
impl ExtensionRegistry {
pub fn new() -> Self {
Self {
handlers: HashMap::new(),
type_urls: HashMap::new(),
descriptors: Vec::new(),
}
}
pub fn add_descriptor(&mut self, bytes: Vec<u8>) {
self.descriptors.push(bytes);
}
pub fn descriptors(&self) -> Vec<&[u8]> {
self.descriptors.iter().map(|b| b.as_slice()).collect()
}
fn register<T>(&mut self, ext_type: ExtensionType) -> Result<(), RegistrationError>
where
T: Extension,
{
let canonical_name = T::name();
let type_url = T::type_url();
let handler: Arc<dyn ExtensionConverter> =
Arc::new(ExtensionAdapter::<T>(std::marker::PhantomData));
let key = (ext_type, canonical_name.to_string());
if self.handlers.contains_key(&key) {
return Err(RegistrationError::DuplicateName {
ext_type,
name: canonical_name.to_string(),
});
}
let type_url_key = (ext_type, type_url.clone());
if let Some(existing) = self.type_urls.get(&type_url_key)
&& existing != canonical_name
{
return Err(RegistrationError::ConflictingTypeUrl {
type_url,
ext_type,
existing_name: existing.clone(),
});
}
self.handlers.insert(key, Arc::clone(&handler));
self.type_urls
.insert(type_url_key, canonical_name.to_string());
Ok(())
}
pub fn register_relation<T>(&mut self) -> Result<(), RegistrationError>
where
T: Extension,
{
self.register::<T>(ExtensionType::Relation)
}
pub fn register_enhancement<T>(&mut self) -> Result<(), RegistrationError>
where
T: Extension,
{
self.register::<T>(ExtensionType::Enhancement)
}
pub fn register_optimization<T>(&mut self) -> Result<(), RegistrationError>
where
T: Extension,
{
self.register::<T>(ExtensionType::Optimization)
}
pub fn parse_extension(
&self,
extension_name: &str,
args: &ExtensionArgs,
) -> Result<Any, ExtensionError> {
self.parse_with_type(ExtensionType::Relation, extension_name, args)
}
pub fn parse_enhancement(
&self,
enhancement_name: &str,
args: &ExtensionArgs,
) -> Result<Any, ExtensionError> {
self.parse_with_type(ExtensionType::Enhancement, enhancement_name, args)
}
pub fn parse_optimization(
&self,
optimization_name: &str,
args: &ExtensionArgs,
) -> Result<Any, ExtensionError> {
self.parse_with_type(ExtensionType::Optimization, optimization_name, args)
}
fn parse_with_type(
&self,
ext_type: ExtensionType,
name: &str,
args: &ExtensionArgs,
) -> Result<Any, ExtensionError> {
let key = (ext_type, name.to_string());
let handler = self
.handlers
.get(&key)
.ok_or_else(|| ExtensionError::NotFound {
name: name.to_string(),
})?;
handler.parse_detail(args)
}
pub fn decode(&self, detail: AnyRef<'_>) -> Result<(String, ExtensionArgs), ExtensionError> {
self.decode_with_type(ExtensionType::Relation, detail)
}
pub fn decode_enhancement(
&self,
detail: AnyRef<'_>,
) -> Result<(String, ExtensionArgs), ExtensionError> {
self.decode_with_type(ExtensionType::Enhancement, detail)
}
pub fn decode_optimization(
&self,
detail: AnyRef<'_>,
) -> Result<(String, ExtensionArgs), ExtensionError> {
self.decode_with_type(ExtensionType::Optimization, detail)
}
fn decode_with_type(
&self,
ext_type: ExtensionType,
detail: AnyRef<'_>,
) -> Result<(String, ExtensionArgs), ExtensionError> {
let type_url_key = (ext_type, detail.type_url.to_string());
let extension_name =
self.type_urls
.get(&type_url_key)
.ok_or_else(|| ExtensionError::NotFound {
name: detail.type_url.to_string(),
})?;
let name_key = (ext_type, extension_name.clone());
let handler = self
.handlers
.get(&name_key)
.ok_or_else(|| ExtensionError::NotFound {
name: extension_name.clone(),
})?;
let args = handler.textify_detail(detail)?;
Ok((extension_name.clone(), args))
}
pub fn extension_names(&self, ext_type: ExtensionType) -> Vec<&str> {
let mut names: Vec<&str> = self
.type_urls
.iter()
.filter_map(|((t, _), name)| {
if *t == ext_type {
Some(name.as_str())
} else {
None
}
})
.collect();
names.sort_unstable();
names.dedup();
names
}
pub fn has_extension(&self, ext_type: ExtensionType, name: &str) -> bool {
self.handlers.contains_key(&(ext_type, name.to_string()))
}
}
impl fmt::Debug for ExtensionRegistry {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let mut keys: Vec<_> = self
.handlers
.keys()
.map(|(t, n)| (format!("{t:?}"), n.as_str()))
.collect();
keys.sort();
f.debug_struct("ExtensionRegistry")
.field("handlers", &keys)
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::extensions::{ExtensionColumn, ExtensionRelationType, ExtensionValue};
struct TestExtension {
path: String,
batch_size: i64,
}
impl AnyConvertible for TestExtension {
fn to_any(&self) -> Result<Any, ExtensionError> {
let json_str = format!(
r#"{{"path":"{}","batch_size":{}}}"#,
self.path, self.batch_size
);
Ok(Any::new(Self::type_url(), json_str.into_bytes()))
}
fn type_url() -> String {
"test.TestExtension".to_string()
}
fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
let json_str = String::from_utf8(any.value.to_vec())
.map_err(|e| ExtensionError::Custom(format!("Invalid UTF-8: {e}")))?;
if json_str.contains("path") && json_str.contains("batch_size") {
Ok(TestExtension {
path: "test.parquet".to_string(),
batch_size: 1024,
})
} else {
Err(ExtensionError::Custom("Missing fields".to_string()))
}
}
}
impl Explainable for TestExtension {
fn name() -> &'static str {
"TestExtension"
}
fn from_args(args: &ExtensionArgs) -> Result<Self, ExtensionError> {
let mut extractor = args.extractor();
let path: String = extractor.expect_named_arg::<&str>("path")?.to_string();
let batch_size: i64 = extractor.expect_named_arg("batch_size")?;
extractor.check_exhausted()?;
Ok(TestExtension {
path: path.to_string(),
batch_size,
})
}
fn to_args(&self) -> Result<ExtensionArgs, ExtensionError> {
let mut args = ExtensionArgs::new(ExtensionRelationType::Leaf);
args.named.insert(
"path".to_string(),
ExtensionValue::String(self.path.clone()),
);
args.named.insert(
"batch_size".to_string(),
ExtensionValue::Integer(self.batch_size),
);
Ok(args)
}
}
#[test]
fn test_extension_registry_basic() {
let mut registry = ExtensionRegistry::new();
assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 0);
assert!(!registry.has_extension(ExtensionType::Relation, "TestExtension"));
registry.register_relation::<TestExtension>().unwrap();
assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 1);
assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
let mut args = ExtensionArgs::new(ExtensionRelationType::Leaf);
args.named.insert(
"path".to_string(),
ExtensionValue::String("data.parquet".to_string()),
);
args.named
.insert("batch_size".to_string(), ExtensionValue::Integer(2048));
let any = registry.parse_extension("TestExtension", &args).unwrap();
assert_eq!(any.type_url, "test.TestExtension");
let any_ref = any.as_ref();
let result = registry.decode(any_ref).unwrap();
assert_eq!(result.0, "TestExtension");
match result.1.named.get("path") {
Some(ExtensionValue::String(s)) => assert_eq!(s, "test.parquet"), _ => panic!("Expected String for path"),
}
}
#[test]
fn test_extension_args() {
let mut args = ExtensionArgs::new(ExtensionRelationType::Leaf);
args.named.insert(
"path".to_string(),
ExtensionValue::String("data/*.parquet".to_string()),
);
args.named
.insert("batch_size".to_string(), ExtensionValue::Integer(1024));
args.positional.push(ExtensionValue::Reference(0));
args.output_columns.push(ExtensionColumn::Named {
name: "col1".to_string(),
type_spec: "i32".to_string(),
});
let mut extractor = args.extractor();
match extractor.get_named_arg("path") {
Some(ExtensionValue::String(s)) => assert_eq!(s, "data/*.parquet"),
_ => panic!("Expected String for path"),
}
match extractor.get_named_arg("batch_size") {
Some(ExtensionValue::Integer(i)) => assert_eq!(*i, 1024),
_ => panic!("Expected Integer for batch_size"),
}
assert!(extractor.check_exhausted().is_ok());
assert_eq!(args.positional.len(), 1);
assert_eq!(args.output_columns.len(), 1);
}
#[test]
fn test_extension_error_cases() {
let registry = ExtensionRegistry::new();
let args = ExtensionArgs::new(ExtensionRelationType::Leaf);
let result = registry.parse_extension("NonExistent", &args);
assert!(matches!(result, Err(ExtensionError::NotFound { .. })));
let args = ExtensionArgs::new(ExtensionRelationType::Leaf);
let mut extractor = args.extractor();
let result = extractor.get_named_arg("missing");
assert!(result.is_none());
assert!(extractor.check_exhausted().is_ok());
let mut args = ExtensionArgs::new(ExtensionRelationType::Leaf);
args.named
.insert("test".to_string(), ExtensionValue::Integer(42));
let mut extractor = args.extractor();
let result = extractor.get_named_arg("test");
match result {
Some(ExtensionValue::Integer(42)) => {} _ => panic!("Expected Integer(42), got {result:?}"),
}
assert!(extractor.check_exhausted().is_ok());
}
struct TestEnhancement {
hint: String,
}
impl AnyConvertible for TestEnhancement {
fn to_any(&self) -> Result<Any, ExtensionError> {
let json_str = format!(r#"{{"hint":"{}"}}"#, self.hint);
Ok(Any::new(Self::type_url(), json_str.into_bytes()))
}
fn type_url() -> String {
"test.TestExtension".to_string()
}
fn from_any<'a>(any: AnyRef<'a>) -> Result<Self, ExtensionError> {
let json_str = String::from_utf8(any.value.to_vec())
.map_err(|e| ExtensionError::Custom(format!("Invalid UTF-8: {e}")))?;
if json_str.contains("hint") {
Ok(TestEnhancement {
hint: "test_hint".to_string(),
})
} else {
Err(ExtensionError::Custom("Missing hint field".to_string()))
}
}
}
impl Explainable for TestEnhancement {
fn name() -> &'static str {
"TestEnhancement"
}
fn from_args(args: &ExtensionArgs) -> Result<Self, ExtensionError> {
let mut extractor = args.extractor();
let hint: String = extractor.expect_named_arg::<&str>("hint")?.to_string();
extractor.check_exhausted()?;
Ok(TestEnhancement { hint })
}
fn to_args(&self) -> Result<ExtensionArgs, ExtensionError> {
let mut args = ExtensionArgs::new(ExtensionRelationType::Leaf);
args.named.insert(
"hint".to_string(),
ExtensionValue::String(self.hint.clone()),
);
Ok(args)
}
}
#[test]
fn test_namespace_separation() {
let mut registry = ExtensionRegistry::new();
registry.register_relation::<TestExtension>().unwrap();
registry.register_enhancement::<TestEnhancement>().unwrap();
assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
assert!(registry.has_extension(ExtensionType::Enhancement, "TestEnhancement"));
assert_eq!(registry.extension_names(ExtensionType::Relation).len(), 1);
assert_eq!(
registry.extension_names(ExtensionType::Enhancement).len(),
1
);
let mut ext_args = ExtensionArgs::new(ExtensionRelationType::Leaf);
ext_args.named.insert(
"path".to_string(),
ExtensionValue::String("data.parquet".to_string()),
);
ext_args
.named
.insert("batch_size".to_string(), ExtensionValue::Integer(2048));
let ext_any = registry
.parse_extension("TestExtension", &ext_args)
.unwrap();
assert_eq!(ext_any.type_url, "test.TestExtension");
let mut enh_args = ExtensionArgs::new(ExtensionRelationType::Leaf);
enh_args.named.insert(
"hint".to_string(),
ExtensionValue::String("optimize".to_string()),
);
let enh_any = registry
.parse_enhancement("TestEnhancement", &enh_args)
.unwrap();
assert_eq!(enh_any.type_url, "test.TestExtension");
let enh_ref = enh_any.as_ref();
let (name, args) = registry.decode_enhancement(enh_ref).unwrap();
assert_eq!(name, "TestEnhancement");
match args.named.get("hint") {
Some(ExtensionValue::String(s)) => assert_eq!(s, "test_hint"), _ => panic!("Expected String for hint"),
}
}
#[test]
fn test_enhancement_duplicate_registration_returns_error() {
let mut registry = ExtensionRegistry::new();
registry.register_enhancement::<TestEnhancement>().unwrap();
let result = registry.register_enhancement::<TestEnhancement>();
assert!(matches!(
result,
Err(RegistrationError::DuplicateName { .. })
));
}
#[test]
fn test_enhancement_not_found_error() {
let registry = ExtensionRegistry::new();
let args = ExtensionArgs::new(ExtensionRelationType::Leaf);
let result = registry.parse_enhancement("NonExistentEnhancement", &args);
assert!(matches!(result, Err(ExtensionError::NotFound { .. })));
}
struct ConflictingExtension;
impl AnyConvertible for ConflictingExtension {
fn to_any(&self) -> Result<Any, ExtensionError> {
Ok(Any::new(Self::type_url(), vec![]))
}
fn type_url() -> String {
"test.TestExtension".to_string()
}
fn from_any<'a>(_any: AnyRef<'a>) -> Result<Self, ExtensionError> {
Ok(ConflictingExtension)
}
}
impl Explainable for ConflictingExtension {
fn name() -> &'static str {
"ConflictingExtension"
}
fn from_args(_args: &ExtensionArgs) -> Result<Self, ExtensionError> {
Ok(ConflictingExtension)
}
fn to_args(&self) -> Result<ExtensionArgs, ExtensionError> {
Ok(ExtensionArgs::new(ExtensionRelationType::Leaf))
}
}
#[test]
fn test_conflicting_type_url_leaves_registry_unchanged() {
let mut registry = ExtensionRegistry::new();
registry.register_relation::<TestExtension>().unwrap();
let result = registry.register_relation::<ConflictingExtension>();
assert!(matches!(
result,
Err(RegistrationError::ConflictingTypeUrl { .. })
));
assert!(registry.has_extension(ExtensionType::Relation, "TestExtension"));
assert!(!registry.has_extension(ExtensionType::Relation, "ConflictingExtension"));
assert_eq!(
registry.extension_names(ExtensionType::Relation),
vec!["TestExtension"]
);
}
}