akar_processor/processor/mapper/
map_ddl.rs

1use super::ExecutionContext;
2use crate::physical_operator::*;
3use crate::processor::SchemaDdlOp;
4use crate::processor::plan_serializer::serialize_plan_tree;
5use akar_common::error::ProcessorError;
6use akar_common::vector::DataChunk;
7use akar_planner::logical_operator::LogicalOperator;
8
9pub fn map_and_execute_ddl(
10    op: &LogicalOperator,
11    current_input: Vec<DataChunk>,
12    ctx: &mut ExecutionContext,
13) -> Result<Vec<DataChunk>, ProcessorError> {
14    match op {
15        LogicalOperator::Explain(ex) => {
16            // Serialize the inner plan tree to a string
17            let plan_str = serialize_plan_tree(&ex.inner, 0);
18            let explain = PhysicalExplain { inner_plan: plan_str };
19            let result = explain.execute(vec![])?;
20            Ok(result)
21        }
22        LogicalOperator::StandaloneCall(c) => {
23            if let Some(ref handler) = ctx.standalone_call_handler {
24                let result = handler.execute_call(&c.function_name, &c.args)?;
25                Ok(result)
26            } else {
27                Err(format!("No standalone call handler available to execute '{}'", c.function_name).into())
28            }
29        }
30        LogicalOperator::TableFunctionCall(tf) => {
31            let result = ctx.processor.execute_table_function(tf)?;
32            Ok(result)
33        }
34        LogicalOperator::Foreach(fc) => {
35            let foreach_op = PhysicalForeach {
36                variable: fc.variable.clone(),
37                expression: fc.expression.clone(),
38                sub_plans: fc.sub_plans.clone(),
39                function_registry: ctx.function_registry.clone(),
40                table_catalog: ctx.table_catalog.clone(),
41                vfs: ctx.vfs.clone(),
42            };
43            let result = foreach_op.execute(current_input)?;
44            Ok(result)
45        }
46        LogicalOperator::CreateNodeTable(c) => {
47            let tc = ctx
48                .table_catalog
49                .as_ref()
50                .ok_or("CREATE NODE TABLE requires a table catalog")?;
51            if c.if_not_exists && tc.get_node_table_by_name(&c.name).is_some() {
52                // IF NOT EXISTS: table already present — idempotent no-op (P72).
53                tracing::info!("CREATE NODE TABLE IF NOT EXISTS: '{}' already exists, skipping", c.name);
54                return Ok(ddl_success_chunk(&format!("Node table '{}' already exists", c.name)));
55            }
56            let columns: Vec<akar_storage::table::ColumnDefinition> = c
57                .columns
58                .iter()
59                .map(|col| akar_storage::table::ColumnDefinition {
60                    name: col.name.clone(),
61                    logical_type: col.logical_type,
62                    is_primary_key: col.is_primary_key,
63                    compression: col.compression,
64                })
65                .collect();
66            tc.create_node_table(c.name.clone(), columns);
67
68            // Auto-create ART index for primary key (matches connection/ddl.rs behavior)
69            if c.columns.iter().any(|col| col.is_primary_key) {
70                let index_name = format!("{}_pk_idx", c.name);
71                tc.create_art_index(&c.name, &index_name)
72                    .map_err(|e| format!("Failed to auto-create ART PK index for table '{}': {e}", c.name))?;
73            }
74
75            tracing::info!("Pipeline: Created node table '{}'", c.name);
76            Ok(ddl_success_chunk(&format!("Node table '{}' created", c.name)))
77        }
78        LogicalOperator::CreateRelTable(c) => {
79            let tc = ctx
80                .table_catalog
81                .as_ref()
82                .ok_or("CREATE REL TABLE requires a table catalog")?;
83            if c.if_not_exists && tc.get_rel_table_by_name(&c.name).is_some() {
84                // IF NOT EXISTS: rel table already present — idempotent no-op (P72).
85                tracing::info!("CREATE REL TABLE IF NOT EXISTS: '{}' already exists, skipping", c.name);
86                return Ok(ddl_success_chunk(&format!("Rel table '{}' already exists", c.name)));
87            }
88            let from_id = tc
89                .get_node_table_by_name(&c.from)
90                .map(|t| t.table_id)
91                .ok_or_else(|| format!("From table '{}' not found", c.from))?;
92            let to_id = tc
93                .get_node_table_by_name(&c.to)
94                .map(|t| t.table_id)
95                .ok_or_else(|| format!("To table '{}' not found", c.to))?;
96            let columns: Vec<akar_storage::table::ColumnDefinition> = c
97                .columns
98                .iter()
99                .map(|col| akar_storage::table::ColumnDefinition {
100                    name: col.name.clone(),
101                    logical_type: col.logical_type,
102                    is_primary_key: col.is_primary_key,
103                    compression: col.compression,
104                })
105                .collect();
106            tc.create_rel_table(c.name.clone(), from_id, to_id, columns);
107            tracing::info!("Pipeline: Created rel table '{}' ({} -> {})", c.name, c.from, c.to);
108            Ok(ddl_success_chunk(&format!("Rel table '{}' created", c.name)))
109        }
110        LogicalOperator::DropTable(d) => {
111            let tc = ctx
112                .table_catalog
113                .as_ref()
114                .ok_or("DROP TABLE requires a table catalog")?;
115            let dropped = tc.drop_node_table(&d.name) || tc.drop_rel_table(&d.name);
116            if dropped {
117                tracing::info!("Pipeline: Dropped table '{}'", d.name);
118                Ok(ddl_success_chunk(&format!("Table '{}' dropped", d.name)))
119            } else {
120                Err(format!("Table '{}' not found", d.name).into())
121            }
122        }
123        LogicalOperator::AlterTable(a) => {
124            let tc = ctx
125                .table_catalog
126                .as_ref()
127                .ok_or("ALTER TABLE requires a table catalog")?;
128            match &a.action {
129                akar_parser::ast::AlterAction::AddColumn { name, type_name } => {
130                    let logical_type = parse_type_simple(type_name)?;
131                    let mut table = tc
132                        .get_node_table_by_name_mut(&a.table_name)
133                        .ok_or_else(|| format!("Table '{}' not found", a.table_name))?;
134                    if table.columns.iter().any(|c| c.name.eq_ignore_ascii_case(name)) {
135                        return Err(format!("Column '{}' already exists in '{}'", name, a.table_name).into());
136                    }
137                    table.columns.push(akar_storage::table::ColumnDefinition {
138                        name: name.clone(),
139                        logical_type,
140                        is_primary_key: false,
141                        compression: akar_common::enums::CompressionType::Uncompressed,
142                    });
143                    tracing::info!("Pipeline: Added column '{}' to '{}'", name, a.table_name);
144                    Ok(ddl_success_chunk(&format!(
145                        "Column '{}' added to table '{}'",
146                        name, a.table_name
147                    )))
148                }
149                akar_parser::ast::AlterAction::DropColumn { name } => {
150                    let mut table = tc
151                        .get_node_table_by_name_mut(&a.table_name)
152                        .ok_or_else(|| format!("Table '{}' not found", a.table_name))?;
153                    let pos = table
154                        .columns
155                        .iter()
156                        .position(|c| c.name == *name)
157                        .ok_or_else(|| format!("Column '{}' not found in '{}'", name, a.table_name))?;
158                    if table.columns[pos].is_primary_key {
159                        return Err(format!("Cannot drop primary key column '{}'", name).into());
160                    }
161                    table.columns.remove(pos);
162                    tracing::info!("Pipeline: Dropped column '{}' from '{}'", name, a.table_name);
163                    Ok(ddl_success_chunk(&format!(
164                        "Column '{}' dropped from table '{}'",
165                        name, a.table_name
166                    )))
167                }
168                akar_parser::ast::AlterAction::RenameColumn { old_name, new_name } => {
169                    {
170                        let table = tc
171                            .get_node_table_by_name(&a.table_name)
172                            .ok_or_else(|| format!("Table '{}' not found", a.table_name))?;
173                        if !table.columns.iter().any(|c| c.name == *old_name) {
174                            return Err(format!("Column '{}' not found in '{}'", old_name, a.table_name).into());
175                        }
176                        if table.columns.iter().any(|c| c.name == *new_name) {
177                            return Err(format!("Column '{}' already exists in '{}'", new_name, a.table_name).into());
178                        }
179                    }
180                    let mut table = tc.get_node_table_by_name_mut(&a.table_name).unwrap();
181                    let col = table.columns.iter_mut().find(|c| c.name == *old_name).unwrap();
182                    col.name = new_name.clone();
183                    tracing::info!(
184                        "Pipeline: Renamed column '{}' to '{}' in '{}'",
185                        old_name,
186                        new_name,
187                        a.table_name
188                    );
189                    Ok(ddl_success_chunk(&format!(
190                        "Column '{}' renamed to '{}' in table '{}'",
191                        old_name, new_name, a.table_name
192                    )))
193                }
194                akar_parser::ast::AlterAction::RenameTable { new_name } => {
195                    if tc.get_node_table_by_name(new_name).is_some() || tc.get_rel_table_by_name(new_name).is_some() {
196                        return Err(format!("Table '{}' already exists", new_name).into());
197                    }
198                    if let Some(mut table) = tc.get_node_table_by_name_mut(&a.table_name) {
199                        table.name = new_name.clone();
200                    } else if let Some(mut table) = tc.get_rel_table_by_name_mut(&a.table_name) {
201                        table.name = new_name.clone();
202                    } else {
203                        return Err(format!("Table '{}' not found", a.table_name).into());
204                    }
205                    tracing::info!("Pipeline: Renamed table '{}' to '{}'", a.table_name, new_name);
206                    Ok(ddl_success_chunk(&format!(
207                        "Table '{}' renamed to '{}'",
208                        a.table_name, new_name
209                    )))
210                }
211            }
212        }
213        LogicalOperator::CreateIndex(idx) => {
214            let tc = ctx
215                .table_catalog
216                .as_ref()
217                .ok_or("CREATE INDEX requires a table catalog")?;
218            tc.create_art_index(&idx.table_name, &idx.index_name)?;
219            tracing::info!(
220                "Pipeline: Created ART index '{}' on '{}'",
221                idx.index_name,
222                idx.table_name
223            );
224            Ok(ddl_success_chunk(&format!(
225                "ART index '{}' created on table '{}'",
226                idx.index_name, idx.table_name
227            )))
228        }
229        LogicalOperator::DropIndex(idx) => {
230            let tc = ctx
231                .table_catalog
232                .as_ref()
233                .ok_or("DROP INDEX requires a table catalog")?;
234            tc.drop_art_index(&idx.table_name)?;
235            tracing::info!("Pipeline: Dropped index '{}' from '{}'", idx.index_name, idx.table_name);
236            Ok(ddl_success_chunk(&format!(
237                "Index '{}' dropped from table '{}'",
238                idx.index_name, idx.table_name
239            )))
240        }
241        LogicalOperator::CreateVectorIndex(vi) => {
242            let tc = ctx
243                .table_catalog
244                .as_ref()
245                .ok_or("CREATE VECTOR INDEX requires a table catalog")?;
246
247            // Map string metric to typed enum
248            let metric = match vi.metric.to_lowercase().as_str() {
249                "cosine" => akar_vector::hnsw::DistanceMetric::Cosine,
250                "euclidean" | "l2" => akar_vector::hnsw::DistanceMetric::L2Squared,
251                "dot" => akar_vector::hnsw::DistanceMetric::DotProduct,
252                other => return Err(format!("Unknown vector metric '{other}'").into()),
253            };
254
255            // Create the vector index in storage
256            tc.create_vector_index(
257                vi.index_name.clone(),
258                vi.table_name.clone(),
259                vi.column_name.clone(),
260                metric,
261                vi.dimensions as u32,
262            );
263
264            // Auto-populate from existing table data
265            if let Some(table) = tc.get_node_table_by_name(&vi.table_name) {
266                let col_idx = table.columns.iter().position(|c| c.name == vi.column_name);
267                if let Some(col_idx) = col_idx {
268                    for row_id in 0..table.num_rows as usize {
269                        if let Some(val) = table.get_value(row_id, col_idx) {
270                            if let Ok(vec) = akar_storage::extract_f64_list_from_value(val) {
271                                if let Some(mut vib) = tc.get_vector_index_by_name_mut(&vi.index_name) {
272                                    vib.hnsw_mut().insert(vec, row_id);
273                                }
274                            }
275                        }
276                    }
277                }
278            }
279
280            tracing::info!(
281                "Pipeline: Created vector index '{}' on '{}.{}'",
282                vi.index_name,
283                vi.table_name,
284                vi.column_name
285            );
286            Ok(ddl_success_chunk(&format!(
287                "Vector index '{}' created on '{}.{}'",
288                vi.index_name, vi.table_name, vi.column_name
289            )))
290        }
291        LogicalOperator::CreateSequence(s) => {
292            if let Some(ref ddl_fn) = ctx.schema_ddl_fn {
293                let result = ddl_fn(SchemaDdlOp::CreateSequence {
294                    name: s.name.clone(),
295                    if_not_exists: s.if_not_exists,
296                    start_value: s.start_with,
297                    increment: s.increment,
298                    min_value: s.min_value,
299                    max_value: s.max_value,
300                    cycle: s.cycle,
301                })?;
302                Ok(ddl_success_chunk(&result))
303            } else {
304                Err("CREATE SEQUENCE requires schema catalog access".into())
305            }
306        }
307        LogicalOperator::DropSequence(s) => {
308            if let Some(ref ddl_fn) = ctx.schema_ddl_fn {
309                let result = ddl_fn(SchemaDdlOp::DropSequence {
310                    name: s.name.clone(),
311                    if_exists: s.if_exists,
312                })?;
313                Ok(ddl_success_chunk(&result))
314            } else {
315                Err("DROP SEQUENCE requires schema catalog access".into())
316            }
317        }
318        LogicalOperator::CreateDml(c) => {
319            let tc = ctx
320                .table_catalog
321                .as_ref()
322                .ok_or("CREATE DML requires a table catalog")?;
323            let mut table = tc
324                .get_node_table_by_name_mut(&c.table_name)
325                .ok_or_else(|| format!("Table '{}' not found", c.table_name))?;
326
327            // Build values from pattern properties, defaulting to Null
328            let mut values: Vec<akar_common::types::Value> =
329                table.columns.iter().map(|_| akar_common::types::Value::Null).collect();
330            {
331                let registry = ctx
332                    .function_registry
333                    .clone()
334                    .ok_or("CREATE DML requires a function registry")?;
335                let registry = registry.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
336                for (prop_name, expr) in &c.properties {
337                    if let Some(col_idx) = table.columns.iter().position(|col| col.name == *prop_name) {
338                        values[col_idx] = crate::physical::write_ops::set::evaluate_constant_expr(expr, &registry);
339                    }
340                }
341            }
342
343            table.insert_row(values)?;
344            tracing::info!("Pipeline: Created node in '{}'", c.table_name);
345            Ok(ddl_success_chunk(&format!("Created node in '{}'", c.table_name)))
346        }
347        LogicalOperator::ExportDatabase(e) => {
348            if let Some(ref ddl_fn) = ctx.schema_ddl_fn {
349                let result = ddl_fn(SchemaDdlOp::ExportDatabase {
350                    file_path: e.file_path.clone(),
351                    file_type: e.file_type.clone(),
352                    schema_only: e.schema_only,
353                })?;
354                Ok(ddl_success_chunk(&result))
355            } else {
356                Err("EXPORT DATABASE requires schema catalog access".into())
357            }
358        }
359        LogicalOperator::ImportDatabase(i) => {
360            if let Some(ref ddl_fn) = ctx.schema_ddl_fn {
361                let result = ddl_fn(SchemaDdlOp::ImportDatabase {
362                    file_path: i.file_path.clone(),
363                    query: i.query.clone(),
364                    index_query: i.index_query.clone(),
365                })?;
366                Ok(ddl_success_chunk(&result))
367            } else {
368                Err("IMPORT DATABASE requires schema catalog access".into())
369            }
370        }
371        LogicalOperator::CreateFtsIndex(c) => {
372            if let Some(ref tc) = ctx.table_catalog {
373                let fts_index = PhysicalCreateFtsIndex {
374                    index_name: c.index_name.clone(),
375                    table_name: c.table_name.clone(),
376                    column_name: c.column_name.clone(),
377                    tokenizer: c.tokenizer.clone(),
378                    if_not_exists: c.if_not_exists,
379                    table_catalog: tc.clone(),
380                };
381                let result = fts_index.execute(current_input)?;
382                Ok(result)
383            } else {
384                Err("CREATE FTS INDEX requires a table catalog".into())
385            }
386        }
387        LogicalOperator::FtsScan(s) => {
388            if let Some(ref tc) = ctx.table_catalog {
389                let fts_scan = PhysicalFtsScan {
390                    index_name: s.index_name.clone(),
391                    query_string: s.query_string.clone(),
392                    table_name: s.table_name.clone(),
393                    column_name: s.column_name.clone(),
394                    table_catalog: tc.clone(),
395                };
396                let result = fts_scan.execute(current_input)?;
397                Ok(result)
398            } else {
399                Err("FTS scan requires a table catalog".into())
400            }
401        }
402        LogicalOperator::EmptyResult(_) => {
403            let exec = crate::physical::misc::PhysicalEmptyResult;
404            let result = exec.execute(current_input)?;
405            Ok(result)
406        }
407        LogicalOperator::MultiplicityReducer(m) => {
408            let exec = crate::physical::misc::PhysicalMultiplicityReducer {
409                key_columns: m.key_columns.clone(),
410            };
411            let input = if !m.children.is_empty() {
412                ctx.execute_children(&m.children)?
413            } else {
414                current_input
415            };
416            let result = exec.execute(input)?;
417            Ok(result)
418        }
419        LogicalOperator::Skip(s) => {
420            let exec = crate::physical::misc::PhysicalSkip {
421                skip_count: s.offset as usize,
422            };
423            let input = if !s.children.is_empty() {
424                ctx.execute_children(&s.children)?
425            } else {
426                current_input
427            };
428            let result = exec.execute(input)?;
429            Ok(result)
430        }
431        LogicalOperator::ExtensionClause(e) => {
432            let exec = crate::physical::misc::PhysicalExtensionClause {
433                action: e.action.clone(),
434                extension_name: e.extension_name.clone(),
435            };
436            let result = exec.execute(current_input)?;
437            Ok(result)
438        }
439        _ => Err(format!("DDL operator not implemented in mapper: {:?}", op).into()),
440    }
441}
442
443/// Build a success DataChunk with a single message column.
444fn ddl_success_chunk(message: &str) -> Vec<DataChunk> {
445    let mut v = akar_common::vector::ValueVector::new(akar_common::types::PhysicalTypeID::String, 1);
446    v.resize(1);
447    v.set_value(0, &akar_common::types::Value::String(message.to_string()))
448        .unwrap();
449    let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
450    let mut chunk = DataChunk::new(vec![arr], vec![akar_common::types::PhysicalTypeID::String]);
451    chunk.size = 1;
452    chunk.field_names = vec!["result".to_string()];
453    vec![chunk]
454}
455
456/// Minimal type parser for ALTER TABLE ADD COLUMN (avoids Akar-binder dependency).
457fn parse_type_simple(type_name: &str) -> Result<akar_common::types::LogicalTypeID, ProcessorError> {
458    let upper = type_name.trim().to_uppercase();
459    match upper.as_str() {
460        "BOOL" | "BOOLEAN" => Ok(akar_common::types::LogicalTypeID::Bool),
461        "INT64" => Ok(akar_common::types::LogicalTypeID::Int64),
462        "INT32" => Ok(akar_common::types::LogicalTypeID::Int32),
463        "INT16" => Ok(akar_common::types::LogicalTypeID::Int16),
464        "INT8" => Ok(akar_common::types::LogicalTypeID::Int8),
465        "UINT64" => Ok(akar_common::types::LogicalTypeID::UInt64),
466        "UINT32" => Ok(akar_common::types::LogicalTypeID::UInt32),
467        "UINT16" => Ok(akar_common::types::LogicalTypeID::UInt16),
468        "UINT8" => Ok(akar_common::types::LogicalTypeID::UInt8),
469        "DOUBLE" => Ok(akar_common::types::LogicalTypeID::Double),
470        "FLOAT" => Ok(akar_common::types::LogicalTypeID::Float),
471        "STRING" => Ok(akar_common::types::LogicalTypeID::String),
472        "BLOB" => Ok(akar_common::types::LogicalTypeID::Blob),
473        "DATE" => Ok(akar_common::types::LogicalTypeID::Date),
474        "TIMESTAMP" => Ok(akar_common::types::LogicalTypeID::Timestamp),
475        "INTERVAL" => Ok(akar_common::types::LogicalTypeID::Interval),
476        "UUID" => Ok(akar_common::types::LogicalTypeID::Uuid),
477        _ => Err(format!("Unknown type '{type_name}'").into()),
478    }
479}
akar_processor/processor/mapper/map_ddl.rs

akar_processor/processor/mapper/
map_ddl.rs
