linkrs_query/executor/streaming/operators/
ddl_operator.rs

1use std::sync::Arc;
2
3use parking_lot::RwLock;
4
5use crate::executor::streaming::chunk::{ColumnInfo, DataChunk, Schema};
6use crate::executor::streaming::executor::StreamingExecutor;
7use crate::executor::streaming::operators::source_operator::OperatorConfig;
8use crate::executor::streaming::operators::spec::{
9    EdgeManageCommand, IndexManageCommand, MacroManageCommand, MigrateAction,
10    SequenceManageCommand, SpaceManageCommand, TagManageCommand, TypeManageCommand,
11    UserManageCommand,
12};
13use crate::executor::streaming::runtime::ExecutionRuntime;
14use crate::executor::streaming::slot::{SlotInfo, SlotLayout};
15use crate::storage::{QueryStorage, StorageSchemaOps};
16use linkrs_core::error::QueryError;
17use linkrs_core::{NullType, Value};
18
19/// Pre-computed layout for DDL manage result chunks (action, name, status).
20fn manage_result_layout() -> Arc<SlotLayout> {
21    use std::sync::OnceLock;
22    static LAYOUT: OnceLock<Arc<SlotLayout>> = OnceLock::new();
23    LAYOUT
24        .get_or_init(|| {
25            Arc::new(SlotLayout::new(vec![
26                SlotInfo {
27                    slot_id: 0,
28                    name: "action".to_string(),
29                    alias: None,
30                    data_type: Some(linkrs_core::DataType::String),
31                    nullable: false,
32                    origin: None,
33                },
34                SlotInfo {
35                    slot_id: 1,
36                    name: "name".to_string(),
37                    alias: None,
38                    data_type: Some(linkrs_core::DataType::String),
39                    nullable: true,
40                    origin: None,
41                },
42                SlotInfo {
43                    slot_id: 2,
44                    name: "status".to_string(),
45                    alias: None,
46                    data_type: Some(linkrs_core::DataType::String),
47                    nullable: false,
48                    origin: None,
49                },
50            ]))
51        })
52        .clone()
53}
54
55mod auth_executor;
56mod maintenance_executor;
57mod migration_executor;
58mod schema_executor;
59
60fn make_manage_result(action: &str, name: Option<&str>, status: &str) -> DataChunk {
61    let name_val = name
62        .map(Value::string)
63        .unwrap_or(Value::Null(NullType::Null));
64    DataChunk::new_with_layout(
65        vec![vec![Value::string(action), name_val, Value::string(status)]],
66        manage_result_layout(),
67    )
68}
69
70/// Multi-row variant of [`make_manage_result`] for batch DDL (e.g.
71/// `DROP TAG a, b, c`): one `(action, name, status)` row per item so each
72/// deletion is reported individually.
73fn make_manage_results(action: &str, items: Vec<(String, &str)>) -> DataChunk {
74    let rows = items
75        .into_iter()
76        .map(|(name, status)| {
77            vec![
78                Value::string(action),
79                Value::string(name),
80                Value::string(status),
81            ]
82        })
83        .collect();
84    DataChunk::new_with_layout(rows, manage_result_layout())
85}
86
87fn exec_ddl<F>(
88    storage: &Option<Arc<RwLock<dyn QueryStorage>>>,
89    f: F,
90) -> Result<Option<DataChunk>, QueryError>
91where
92    F: FnOnce(&mut dyn StorageSchemaOps) -> Result<(), QueryError>,
93{
94    if let Some(lock) = storage {
95        let mut writer = lock.write();
96        f(&mut *writer).map(|_| Some(make_manage_result("ddl", None, "executed")))
97    } else {
98        Ok(Some(make_manage_result("ddl", None, "no-storage")))
99    }
100}
101
102fn exec_auth<F>(
103    storage: &Option<Arc<RwLock<dyn QueryStorage>>>,
104    f: F,
105) -> Result<Option<DataChunk>, QueryError>
106where
107    F: FnOnce(&mut dyn QueryStorage) -> Result<(), QueryError>,
108{
109    if let Some(lock) = storage {
110        let mut writer = lock.write();
111        f(&mut *writer).map(|_| Some(make_manage_result("auth", None, "executed")))
112    } else {
113        Ok(Some(make_manage_result("auth", None, "no-storage")))
114    }
115}
116
117fn get_reader(
118    storage: &Option<Arc<RwLock<dyn QueryStorage>>>,
119) -> Result<parking_lot::RwLockReadGuard<'_, dyn QueryStorage>, QueryError> {
120    storage
121        .as_ref()
122        .map(|s| s.read())
123        .ok_or_else(|| QueryError::execution("No storage available".to_string()))
124}
125
126pub(super) fn make_single_row(schema: Arc<Schema>, cols: Vec<Value>) -> DataChunk {
127    DataChunk::new(vec![cols], schema)
128}
129
130fn make_single_col_schema(col_name: &str, col_type: &str) -> Arc<Schema> {
131    Arc::new(Schema::new(vec![ColumnInfo {
132        name: col_name.to_string(),
133        data_type: col_type.to_string(),
134    }]))
135}
136
137#[derive(Debug)]
138pub enum DdlOperatorKind {
139    SpaceManage {
140        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
141        command: SpaceManageCommand,
142        emitted: bool,
143    },
144    TagManage {
145        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
146        space_name: String,
147        command: TagManageCommand,
148        emitted: bool,
149    },
150    EdgeManage {
151        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
152        space_name: String,
153        command: EdgeManageCommand,
154        emitted: bool,
155    },
156    IndexManage {
157        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
158        space_name: String,
159        command: IndexManageCommand,
160        emitted: bool,
161    },
162    DeleteIndex {
163        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
164        space_name: String,
165        index_name: String,
166        emitted: bool,
167    },
168    UserManage {
169        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
170        command: UserManageCommand,
171        emitted: bool,
172    },
173    ShowStats {
174        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
175        space_name: String,
176        emitted: bool,
177    },
178    ShowConfigs {
179        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
180        space_name: String,
181        emitted: bool,
182    },
183    ShowQueries {
184        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
185        space_name: String,
186        emitted: bool,
187    },
188    ShowSessions {
189        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
190        space_name: String,
191        emitted: bool,
192    },
193    ShowFunctions {
194        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
195        space_name: String,
196        emitted: bool,
197    },
198    ShowGraphs {
199        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
200        space_name: String,
201        emitted: bool,
202    },
203    ShowMacros {
204        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
205        space_name: String,
206        emitted: bool,
207    },
208    ShowAttachedDatabases {
209        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
210        space_name: String,
211        emitted: bool,
212    },
213    ShowExtensions {
214        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
215        space_name: String,
216        emitted: bool,
217    },
218    LoadFrom {
219        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
220        space_name: String,
221        source_kind: String,
222        source_value: String,
223        func_name: Option<String>,
224        func_args_json: Option<String>,
225        options: Vec<(String, String)>,
226        col_names: Vec<String>,
227        headers: Vec<String>,
228        emitted: bool,
229    },
230    InQueryCall {
231        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
232        space_name: String,
233        func_name: String,
234        args_json: String,
235        yield_items: Vec<(String, String)>,
236        col_names: Vec<String>,
237        emitted: bool,
238    },
239    Analyze {
240        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
241        space_name: String,
242        analyze_target: String,
243        target_name: Option<String>,
244        emitted: bool,
245    },
246    Migrate {
247        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
248        space_name: String,
249        action: MigrateAction,
250        migration_data: Option<String>,
251        emitted: bool,
252    },
253    MigratePlan {
254        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
255        space_name: String,
256        label: String,
257        is_edge: bool,
258        from_version: u64,
259        to_version: u64,
260        emitted: bool,
261    },
262    MigrateRun {
263        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
264        plan_json: String,
265        emitted: bool,
266    },
267    MigrateRollback {
268        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
269        plan_json: String,
270        emitted: bool,
271    },
272    SequenceManage {
273        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
274        command: SequenceManageCommand,
275        emitted: bool,
276    },
277    MacroManage {
278        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
279        command: MacroManageCommand,
280        emitted: bool,
281    },
282    TypeManage {
283        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
284        command: TypeManageCommand,
285        emitted: bool,
286    },
287}
288
289/// DDL operator.
290///
291/// Wraps [`DdlOperatorKind`] with the runtime context injected at `open()`.
292/// Lifecycle state is owned exclusively by the executor; operators never
293/// write it.
294#[derive(Debug)]
295pub struct DdlOperator {
296    pub kind: DdlOperatorKind,
297    pub runtime: Option<Arc<ExecutionRuntime>>,
298    pub output_layout: Arc<SlotLayout>,
299    pub config: OperatorConfig,
300}
301
302impl DdlOperator {
303    pub fn from_spec(
304        spec: &super::spec::DdlSpec,
305        storage: Option<Arc<RwLock<dyn QueryStorage>>>,
306        output_layout: Arc<SlotLayout>,
307    ) -> Self {
308        let kind = match spec {
309            super::spec::DdlSpec::SpaceManage { command } => DdlOperatorKind::SpaceManage {
310                storage: storage.clone(),
311                command: command.clone(),
312                emitted: false,
313            },
314            super::spec::DdlSpec::TagManage {
315                space_name,
316                command,
317            } => DdlOperatorKind::TagManage {
318                storage: storage.clone(),
319                space_name: space_name.clone(),
320                command: command.clone(),
321                emitted: false,
322            },
323            super::spec::DdlSpec::EdgeManage {
324                space_name,
325                command,
326            } => DdlOperatorKind::EdgeManage {
327                storage: storage.clone(),
328                space_name: space_name.clone(),
329                command: command.clone(),
330                emitted: false,
331            },
332            super::spec::DdlSpec::IndexManage {
333                space_name,
334                command,
335            } => DdlOperatorKind::IndexManage {
336                storage: storage.clone(),
337                space_name: space_name.clone(),
338                command: command.clone(),
339                emitted: false,
340            },
341            super::spec::DdlSpec::DeleteIndex {
342                space_name,
343                index_name,
344            } => DdlOperatorKind::DeleteIndex {
345                storage: storage.clone(),
346                space_name: space_name.clone(),
347                index_name: index_name.clone(),
348                emitted: false,
349            },
350            super::spec::DdlSpec::UserManage { command } => DdlOperatorKind::UserManage {
351                storage: storage.clone(),
352                command: command.clone(),
353                emitted: false,
354            },
355            super::spec::DdlSpec::ShowStats { space_name } => DdlOperatorKind::ShowStats {
356                storage: storage.clone(),
357                space_name: space_name.clone(),
358                emitted: false,
359            },
360            super::spec::DdlSpec::ShowConfigs { space_name } => DdlOperatorKind::ShowConfigs {
361                storage: storage.clone(),
362                space_name: space_name.clone(),
363                emitted: false,
364            },
365            super::spec::DdlSpec::ShowQueries { space_name } => DdlOperatorKind::ShowQueries {
366                storage: storage.clone(),
367                space_name: space_name.clone(),
368                emitted: false,
369            },
370            super::spec::DdlSpec::ShowSessions { space_name } => DdlOperatorKind::ShowSessions {
371                storage: storage.clone(),
372                space_name: space_name.clone(),
373                emitted: false,
374            },
375            super::spec::DdlSpec::ShowFunctions { space_name } => DdlOperatorKind::ShowFunctions {
376                storage: storage.clone(),
377                space_name: space_name.clone(),
378                emitted: false,
379            },
380            super::spec::DdlSpec::ShowGraphs { space_name } => DdlOperatorKind::ShowGraphs {
381                storage: storage.clone(),
382                space_name: space_name.clone(),
383                emitted: false,
384            },
385            super::spec::DdlSpec::ShowMacros { space_name } => DdlOperatorKind::ShowMacros {
386                storage: storage.clone(),
387                space_name: space_name.clone(),
388                emitted: false,
389            },
390            super::spec::DdlSpec::ShowAttachedDatabases { space_name } => {
391                DdlOperatorKind::ShowAttachedDatabases {
392                    storage: storage.clone(),
393                    space_name: space_name.clone(),
394                    emitted: false,
395                }
396            }
397            super::spec::DdlSpec::ShowExtensions { space_name } => {
398                DdlOperatorKind::ShowExtensions {
399                    storage: storage.clone(),
400                    space_name: space_name.clone(),
401                    emitted: false,
402                }
403            }
404            super::spec::DdlSpec::LoadFrom {
405                space_name,
406                source_kind,
407                source_value,
408                func_name,
409                func_args_json,
410                options,
411                col_names,
412                headers,
413            } => DdlOperatorKind::LoadFrom {
414                storage: storage.clone(),
415                space_name: space_name.clone(),
416                source_kind: source_kind.clone(),
417                source_value: source_value.clone(),
418                func_name: func_name.clone(),
419                func_args_json: func_args_json.clone(),
420                options: options.clone(),
421                col_names: col_names.clone(),
422                headers: headers.clone(),
423                emitted: false,
424            },
425            super::spec::DdlSpec::InQueryCall {
426                space_name,
427                func_name,
428                args_json,
429                yield_items,
430                col_names,
431            } => DdlOperatorKind::InQueryCall {
432                storage: storage.clone(),
433                space_name: space_name.clone(),
434                func_name: func_name.clone(),
435                args_json: args_json.clone(),
436                yield_items: yield_items.clone(),
437                col_names: col_names.clone(),
438                emitted: false,
439            },
440            super::spec::DdlSpec::Analyze { space_name } => DdlOperatorKind::Analyze {
441                storage: storage.clone(),
442                space_name: space_name.clone(),
443                analyze_target: String::new(),
444                target_name: None,
445                emitted: false,
446            },
447            super::spec::DdlSpec::Migrate {
448                space_name,
449                action,
450                migration_data,
451            } => DdlOperatorKind::Migrate {
452                storage,
453                space_name: space_name.clone(),
454                action: *action,
455                migration_data: migration_data.clone(),
456                emitted: false,
457            },
458            super::spec::DdlSpec::MigratePlan {
459                space_name,
460                label,
461                is_edge,
462                from_version,
463                to_version,
464            } => DdlOperatorKind::MigratePlan {
465                storage: storage.clone(),
466                space_name: space_name.clone(),
467                label: label.clone(),
468                is_edge: *is_edge,
469                from_version: *from_version,
470                to_version: *to_version,
471                emitted: false,
472            },
473            super::spec::DdlSpec::MigrateRun { plan_json } => DdlOperatorKind::MigrateRun {
474                storage: storage.clone(),
475                plan_json: plan_json.clone(),
476                emitted: false,
477            },
478            super::spec::DdlSpec::MigrateRollback { plan_json } => {
479                DdlOperatorKind::MigrateRollback {
480                    storage: storage.clone(),
481                    plan_json: plan_json.clone(),
482                    emitted: false,
483                }
484            }
485            super::spec::DdlSpec::SequenceManage { command } => DdlOperatorKind::SequenceManage {
486                storage,
487                command: command.clone(),
488                emitted: false,
489            },
490            super::spec::DdlSpec::MacroManage { command } => DdlOperatorKind::MacroManage {
491                storage,
492                command: command.clone(),
493                emitted: false,
494            },
495            super::spec::DdlSpec::TypeManage { command } => DdlOperatorKind::TypeManage {
496                storage,
497                command: command.clone(),
498                emitted: false,
499            },
500        };
501        Self::new(kind, output_layout)
502    }
503
504    pub fn new(kind: DdlOperatorKind, output_layout: Arc<SlotLayout>) -> Self {
505        Self {
506            kind,
507            runtime: None,
508            output_layout,
509            config: OperatorConfig::default(),
510        }
511    }
512
513    /// Inject the runtime and execution config (called once by the executor
514    /// before this operator produces any data).
515    pub fn inject_context(
516        &mut self,
517        runtime: Option<&Arc<ExecutionRuntime>>,
518        config: OperatorConfig,
519    ) {
520        if let Some(rt) = runtime {
521            self.runtime = Some(rt.clone());
522        }
523        self.config = config;
524    }
525
526    pub fn open(&mut self, input: &mut StreamingExecutor) -> Result<(), QueryError> {
527        input.open()?;
528        Ok(())
529    }
530
531    pub fn next(
532        &mut self,
533        _input: &mut StreamingExecutor,
534    ) -> Result<Option<DataChunk>, QueryError> {
535        match &mut self.kind {
536            DdlOperatorKind::SpaceManage { .. } => schema_executor::execute_space_manage(self),
537            DdlOperatorKind::TagManage { .. } => schema_executor::execute_tag_manage(self),
538            DdlOperatorKind::EdgeManage { .. } => schema_executor::execute_edge_manage(self),
539            DdlOperatorKind::IndexManage { .. } => schema_executor::execute_index_manage(self),
540            DdlOperatorKind::DeleteIndex { .. } => schema_executor::execute_delete_index(self),
541            DdlOperatorKind::UserManage { .. } => auth_executor::execute_user_manage(self),
542            DdlOperatorKind::ShowStats { .. } => maintenance_executor::execute_show_stats(self),
543            DdlOperatorKind::ShowConfigs { .. } => maintenance_executor::execute_show_configs(self),
544            DdlOperatorKind::ShowQueries { .. } => maintenance_executor::execute_show_queries(self),
545            DdlOperatorKind::ShowSessions { .. } => {
546                maintenance_executor::execute_show_sessions(self)
547            }
548            DdlOperatorKind::ShowFunctions { .. } => {
549                maintenance_executor::execute_show_functions(self)
550            }
551            DdlOperatorKind::ShowGraphs { .. } => maintenance_executor::execute_show_graphs(self),
552            DdlOperatorKind::ShowMacros { .. } => maintenance_executor::execute_show_macros(self),
553            DdlOperatorKind::ShowAttachedDatabases { .. } => {
554                maintenance_executor::execute_show_attached_databases(self)
555            }
556            DdlOperatorKind::ShowExtensions { .. } => {
557                maintenance_executor::execute_show_extensions(self)
558            }
559            DdlOperatorKind::LoadFrom { .. } => maintenance_executor::execute_load_from(self),
560            DdlOperatorKind::InQueryCall { .. } => {
561                maintenance_executor::execute_in_query_call(self)
562            }
563            DdlOperatorKind::Analyze { .. } => maintenance_executor::execute_analyze(self),
564            DdlOperatorKind::Migrate { .. } => maintenance_executor::execute_migrate(self),
565            DdlOperatorKind::MigratePlan { .. } => migration_executor::execute_migrate_plan(self),
566            DdlOperatorKind::MigrateRun { .. } => migration_executor::execute_migrate_run(self),
567            DdlOperatorKind::MigrateRollback { .. } => {
568                migration_executor::execute_migrate_rollback(self)
569            }
570            DdlOperatorKind::SequenceManage { .. } => self.execute_sequence_manage(),
571            DdlOperatorKind::MacroManage { .. } => self.execute_macro_manage(),
572            DdlOperatorKind::TypeManage { .. } => self.execute_type_manage(),
573        }
574    }
575
576    pub fn stop(&mut self) -> Result<(), QueryError> {
577        Ok(())
578    }
579
580    pub fn close(&mut self) -> Result<(), QueryError> {
581        Ok(())
582    }
583
584    fn execute_sequence_manage(&mut self) -> Result<Option<DataChunk>, QueryError> {
585        let command = match &self.kind {
586            DdlOperatorKind::SequenceManage {
587                command, emitted, ..
588            } => {
589                if *emitted {
590                    return Ok(None);
591                }
592                command.clone()
593            }
594            _ => {
595                return Err(QueryError::execution(
596                    "execute_sequence_manage called with non-SequenceManage kind".to_string(),
597                ));
598            }
599        };
600
601        let manager = self
602            .runtime
603            .as_ref()
604            .and_then(|rt| rt.sequence_manager())
605            .ok_or_else(|| {
606                QueryError::execution(
607                    "CREATE/ALTER/DROP SEQUENCE requires a sequence catalog manager on the runtime"
608                        .to_string(),
609                )
610            })?;
611
612        let result = match command {
613            SequenceManageCommand::Create {
614                seq_name,
615                start,
616                increment,
617                min_value,
618                max_value,
619                cycle,
620                if_not_exists,
621            } => {
622                if manager.exists(&seq_name) {
623                    if if_not_exists {
624                        Ok(Some(make_manage_result(
625                            "create",
626                            Some(&seq_name),
627                            "if_not_exists",
628                        )))
629                    } else {
630                        Err(QueryError::execution(format!(
631                            "Sequence '{seq_name}' already exists"
632                        )))
633                    }
634                } else {
635                    let increment = increment.unwrap_or(1);
636                    if increment == 0 {
637                        return Err(QueryError::execution(format!(
638                            "Sequence '{seq_name}' has INCREMENT=0, which would never advance"
639                        )));
640                    }
641                    manager
642                        .create_sequence(
643                            seq_name.clone(),
644                            start.unwrap_or(1),
645                            increment,
646                            min_value.unwrap_or(i64::MIN),
647                            max_value.unwrap_or(i64::MAX),
648                            cycle,
649                        )
650                        .map_err(|e| QueryError::execution(e.to_string()))?;
651                    Ok(Some(make_manage_result("create", Some(&seq_name), "ok")))
652                }
653            }
654            SequenceManageCommand::Alter {
655                seq_name,
656                increment,
657                min_value,
658                max_value,
659                cycle,
660            } => {
661                if !manager.exists(&seq_name) {
662                    Err(QueryError::execution(format!(
663                        "Sequence '{seq_name}' does not exist"
664                    )))
665                } else {
666                    if increment == Some(0) {
667                        return Err(QueryError::execution(format!(
668                            "Sequence '{seq_name}' has INCREMENT=0, which would never advance"
669                        )));
670                    }
671                    manager
672                        .alter_sequence(&seq_name, increment, min_value, max_value, cycle)
673                        .map_err(|e| QueryError::execution(e.to_string()))?;
674                    Ok(Some(make_manage_result("alter", Some(&seq_name), "ok")))
675                }
676            }
677            SequenceManageCommand::Drop {
678                seq_name,
679                if_exists,
680            } => {
681                if !manager.exists(&seq_name) {
682                    if if_exists {
683                        Ok(Some(make_manage_result(
684                            "drop",
685                            Some(&seq_name),
686                            "if_exists",
687                        )))
688                    } else {
689                        Err(QueryError::execution(format!(
690                            "Sequence '{seq_name}' does not exist"
691                        )))
692                    }
693                } else {
694                    manager
695                        .drop_sequence(&seq_name)
696                        .map_err(|e| QueryError::execution(e.to_string()))?;
697                    Ok(Some(make_manage_result("drop", Some(&seq_name), "ok")))
698                }
699            }
700        };
701        if result.is_ok() {
702            if let DdlOperatorKind::SequenceManage { emitted, .. } = &mut self.kind {
703                *emitted = true;
704            }
705        }
706        result
707    }
708
709    fn execute_macro_manage(&mut self) -> Result<Option<DataChunk>, QueryError> {
710        use linkrs_core::metadata::MacroDef;
711
712        let command = match &self.kind {
713            DdlOperatorKind::MacroManage {
714                command, emitted, ..
715            } => {
716                if *emitted {
717                    return Ok(None);
718                }
719                command.clone()
720            }
721            _ => {
722                return Err(QueryError::execution(
723                    "execute_macro_manage called with non-MacroManage kind".to_string(),
724                ));
725            }
726        };
727
728        let manager = self
729            .runtime
730            .as_ref()
731            .and_then(|rt| rt.macro_manager())
732            .ok_or_else(|| {
733                QueryError::execution(
734                    "CREATE/DROP MACRO requires a macro catalog manager on the runtime".to_string(),
735                )
736            })?;
737
738        let result = match command {
739            MacroManageCommand::Create {
740                macro_name,
741                params,
742                body,
743                if_not_exists,
744            } => {
745                if manager.exists(&macro_name) {
746                    if if_not_exists {
747                        Ok(Some(make_manage_result(
748                            "create",
749                            Some(&macro_name),
750                            "if_not_exists",
751                        )))
752                    } else {
753                        Err(QueryError::execution(format!(
754                            "Macro '{macro_name}' already exists"
755                        )))
756                    }
757                } else {
758                    manager
759                        .create_macro(MacroDef::new(macro_name.clone(), params, body))
760                        .map_err(|e| QueryError::execution(e.to_string()))?;
761                    Ok(Some(make_manage_result("create", Some(&macro_name), "ok")))
762                }
763            }
764            MacroManageCommand::Drop {
765                macro_name,
766                if_exists,
767            } => {
768                if !manager.exists(&macro_name) {
769                    if if_exists {
770                        Ok(Some(make_manage_result(
771                            "drop",
772                            Some(&macro_name),
773                            "if_exists",
774                        )))
775                    } else {
776                        Err(QueryError::execution(format!(
777                            "Macro '{macro_name}' does not exist"
778                        )))
779                    }
780                } else {
781                    manager
782                        .drop_macro(&macro_name)
783                        .map_err(|e| QueryError::execution(e.to_string()))?;
784                    Ok(Some(make_manage_result("drop", Some(&macro_name), "ok")))
785                }
786            }
787        };
788        if result.is_ok() {
789            if let DdlOperatorKind::MacroManage { emitted, .. } = &mut self.kind {
790                *emitted = true;
791            }
792        }
793        result
794    }
795
796    fn execute_type_manage(&mut self) -> Result<Option<DataChunk>, QueryError> {
797        use linkrs_core::metadata::TypeAliasDef;
798
799        let command = match &self.kind {
800            DdlOperatorKind::TypeManage {
801                command, emitted, ..
802            } => {
803                if *emitted {
804                    return Ok(None);
805                }
806                command.clone()
807            }
808            _ => {
809                return Err(QueryError::execution(
810                    "execute_type_manage called with non-TypeManage kind".to_string(),
811                ));
812            }
813        };
814
815        let manager = self
816            .runtime
817            .as_ref()
818            .and_then(|rt| rt.type_alias_manager())
819            .ok_or_else(|| {
820                QueryError::execution(
821                    "CREATE/DROP TYPE requires a type-alias catalog manager on the runtime"
822                        .to_string(),
823                )
824            })?;
825
826        let result = match command {
827            TypeManageCommand::Create {
828                type_name,
829                underlying_type,
830                if_not_exists,
831            } => {
832                if manager.exists(&type_name) {
833                    if if_not_exists {
834                        Ok(Some(make_manage_result(
835                            "create",
836                            Some(&type_name),
837                            "if_not_exists",
838                        )))
839                    } else {
840                        Err(QueryError::execution(format!(
841                            "Type '{type_name}' already exists"
842                        )))
843                    }
844                } else {
845                    // Cycle detection runs inside the catalog write.
846                    manager
847                        .create_alias(TypeAliasDef::new(type_name.clone(), underlying_type))
848                        .map_err(|e| QueryError::execution(e.to_string()))?;
849                    Ok(Some(make_manage_result("create", Some(&type_name), "ok")))
850                }
851            }
852            TypeManageCommand::Drop {
853                type_name,
854                if_exists,
855            } => {
856                if !manager.exists(&type_name) {
857                    if if_exists {
858                        Ok(Some(make_manage_result(
859                            "drop",
860                            Some(&type_name),
861                            "if_exists",
862                        )))
863                    } else {
864                        Err(QueryError::execution(format!(
865                            "Type '{type_name}' does not exist"
866                        )))
867                    }
868                } else {
869                    manager
870                        .drop_alias(&type_name)
871                        .map_err(|e| QueryError::execution(e.to_string()))?;
872                    Ok(Some(make_manage_result("drop", Some(&type_name), "ok")))
873                }
874            }
875        };
876        if result.is_ok() {
877            if let DdlOperatorKind::TypeManage { emitted, .. } = &mut self.kind {
878                *emitted = true;
879            }
880        }
881        result
882    }
883}
linkrs_query/executor/streaming/operators/ddl_operator.rs

linkrs_query/executor/streaming/operators/
ddl_operator.rs
