(
ctx: &SessionContext,
catalogs: &HashMap<String, Arc<dyn Catalog>>,
current_catalog: &str,
func: &Function,
)
| 125 | } |
| 126 | |
| 127 | pub async fn execute_call( |
| 128 | ctx: &SessionContext, |
| 129 | catalogs: &HashMap<String, Arc<dyn Catalog>>, |
| 130 | current_catalog: &str, |
| 131 | func: &Function, |
| 132 | ) -> DFResult<DataFrame> { |
| 133 | let (explicit_catalog, proc_name) = extract_procedure_name(&func.name)?; |
| 134 | let catalog_name = explicit_catalog.as_deref().unwrap_or(current_catalog); |
| 135 | let catalog = catalogs |
| 136 | .get(catalog_name) |
| 137 | .ok_or_else(|| DataFusionError::Plan(format!("Unknown catalog '{catalog_name}'")))?; |
| 138 | let args = extract_named_args(&func.args)?; |
| 139 | |
| 140 | match proc_name.as_str() { |
| 141 | "create_tag" => proc_create_tag(ctx, catalog, catalog_name, &args).await, |
| 142 | "delete_tag" => proc_delete_tag(ctx, catalog, catalog_name, &args).await, |
| 143 | "rollback_to" => proc_rollback_to(ctx, catalog, catalog_name, &args).await, |
| 144 | "rollback_to_timestamp" => { |
| 145 | proc_rollback_to_timestamp(ctx, catalog, catalog_name, &args).await |
| 146 | } |
| 147 | "create_tag_from_timestamp" => { |
| 148 | proc_create_tag_from_timestamp(ctx, catalog, catalog_name, &args).await |
| 149 | } |
| 150 | _ => Err(DataFusionError::Plan(format!( |
| 151 | "Unknown procedure: {proc_name}" |
| 152 | ))), |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | /// Returns (optional_catalog_name, procedure_name). |
| 157 | fn extract_procedure_name(name: &ObjectName) -> DFResult<(Option<String>, String)> { |
no test coverage detected