cqlite_core/schema/aggregator/
mod.rs

1//! Schema Aggregator for M2-CLI
2//!
3//! This module implements schema loading and merging from multiple sources (CQL and JSON files/directories).
4//! It handles two-pass loading (UDTs first, then tables) and implements last-wins merging strategy.
5//!
6//! Responsibilities are split across submodules:
7//! - [`json`] — JSON schema parsing/conversion (minimal + full formats).
8//! - [`cql`] — CQL file parsing (`CREATE TYPE`/`CREATE TABLE`, keyspace context).
9//! - this module — the shared model plus orchestration (file discovery,
10//!   per-file dispatch, two-pass registry application).
11
12mod cql;
13mod json;
14
15use std::collections::HashMap;
16use std::path::{Path, PathBuf};
17use std::sync::Arc;
18use tokio::sync::RwLock;
19
20use crate::error::{Error, Result};
21use crate::schema::{TableSchema, UdtRegistry};
22use crate::types::UdtTypeDef;
23
24#[allow(unused_imports)]
25use crate::schema::cql_parser;
26
27/// Configuration for schema aggregator behavior
28#[derive(Debug, Clone)]
29pub struct AggregatorConfig {
30    /// Whether to continue loading after encountering errors
31    pub graceful_degradation: bool,
32    /// Whether to validate UDT dependencies
33    pub validate_udt_dependencies: bool,
34}
35
36impl Default for AggregatorConfig {
37    fn default() -> Self {
38        Self {
39            graceful_degradation: true,
40            validate_udt_dependencies: true,
41        }
42    }
43}
44
45/// Schema aggregator for loading and merging schemas from multiple sources
46pub struct SchemaAggregator {
47    /// Schema registry for storing table schemas
48    registry: Arc<RwLock<crate::schema::registry::SchemaRegistry>>,
49    /// UDT registry for managing user-defined types
50    udt_registry: Arc<RwLock<UdtRegistry>>,
51    /// Configuration
52    config: AggregatorConfig,
53    /// Collected errors during loading
54    errors: Vec<SchemaLoadError>,
55    /// Collected warnings during loading
56    warnings: Vec<SchemaLoadWarning>,
57}
58
59/// Result of schema loading operation
60#[derive(Debug, Clone)]
61pub struct LoadResult {
62    /// Number of table schemas successfully loaded
63    pub schemas_loaded: usize,
64    /// Number of UDTs successfully loaded
65    pub udts_loaded: usize,
66    /// Errors encountered during loading
67    pub errors: Vec<SchemaLoadError>,
68    /// Warnings encountered during loading
69    pub warnings: Vec<SchemaLoadWarning>,
70}
71
72/// Error encountered during schema loading
73#[derive(Debug, Clone)]
74pub struct SchemaLoadError {
75    /// File path where the error occurred
76    pub file_path: Option<PathBuf>,
77    /// Type of error
78    pub error_type: LoadErrorType,
79    /// Error message
80    pub message: String,
81}
82
83/// Types of schema load errors
84#[derive(Debug, Clone)]
85pub enum LoadErrorType {
86    /// Failed to read file
87    FileRead,
88    /// Invalid JSON format
89    InvalidJson,
90    /// Invalid CQL syntax
91    InvalidCql,
92    /// Missing UDT dependency
93    MissingUdtDependency,
94    /// Circular UDT dependency
95    CircularUdtDependency,
96    /// Schema validation failed
97    ValidationFailed,
98    /// Invalid file format (neither .cql nor .json)
99    InvalidFileFormat,
100}
101
102/// Warning encountered during schema loading
103#[derive(Debug, Clone)]
104pub struct SchemaLoadWarning {
105    /// File path where the warning occurred
106    pub file_path: Option<PathBuf>,
107    /// Warning message
108    pub message: String,
109}
110
111/// Intermediate parsed schema data before registry insertion
112#[derive(Debug, Clone)]
113struct ParsedSchema {
114    /// Keyspace name (for context only; tables/udts are now keyed by qualified names)
115    #[allow(dead_code)]
116    keyspace: String,
117    /// Table schemas (keyed by qualified name: "keyspace.table")
118    tables: HashMap<String, TableSchema>,
119    /// UDT definitions (keyed by qualified name: "keyspace.typename")
120    udts: HashMap<String, UdtTypeDef>,
121}
122
123impl SchemaAggregator {
124    /// Create a new schema aggregator
125    pub fn new(
126        registry: Arc<RwLock<crate::schema::registry::SchemaRegistry>>,
127        udt_registry: Arc<RwLock<UdtRegistry>>,
128        config: AggregatorConfig,
129    ) -> Self {
130        Self {
131            registry,
132            udt_registry,
133            config,
134            errors: Vec::new(),
135            warnings: Vec::new(),
136        }
137    }
138
139    /// Load schemas from multiple paths (files or directories)
140    pub async fn load_from_paths(&mut self, paths: &[PathBuf]) -> Result<LoadResult> {
141        self.errors.clear();
142        self.warnings.clear();
143
144        // Step 1: Discover all files from paths in order
145        let mut all_files = Vec::new();
146        for path in paths {
147            if let Err(e) = self.discover_files(path, &mut all_files) {
148                self.errors.push(SchemaLoadError {
149                    file_path: Some(path.clone()),
150                    error_type: LoadErrorType::FileRead,
151                    message: format!("Failed to discover files: {}", e),
152                });
153            }
154        }
155
156        if all_files.is_empty() && !self.errors.is_empty() {
157            return Ok(self.build_result(0, 0));
158        }
159
160        // Step 2: Parse all files into intermediate format
161        let mut parsed_schemas = Vec::new();
162        for file_path in &all_files {
163            match self.parse_file(file_path).await {
164                Ok(Some(schema)) => parsed_schemas.push(schema),
165                Ok(None) => {} // Skipped file
166                Err(e) => {
167                    // Map error type based on the actual error variant
168                    let error_type = match &e {
169                        Error::Io(_) => LoadErrorType::FileRead,
170                        Error::CqlParse(_) => LoadErrorType::InvalidCql,
171                        Error::Schema(_) => {
172                            // Schema errors are structural validation failures
173                            // Check message for JSON vs general validation
174                            let msg = e.to_string();
175                            if msg.contains("Invalid JSON")
176                                || msg.contains("JSON")
177                                || msg.contains("json")
178                            {
179                                LoadErrorType::InvalidJson
180                            } else {
181                                // Missing partition_keys, bad clustering config, etc.
182                                LoadErrorType::ValidationFailed
183                            }
184                        }
185                        _ => {
186                            // Fallback: check error message for clues
187                            let msg = e.to_string();
188                            if msg.contains("JSON") || msg.contains("json") {
189                                LoadErrorType::InvalidJson
190                            } else if msg.contains("CQL") || msg.contains("parse") {
191                                LoadErrorType::InvalidCql
192                            } else {
193                                // Unknown error types default to validation failure, not I/O
194                                LoadErrorType::ValidationFailed
195                            }
196                        }
197                    };
198                    self.errors.push(SchemaLoadError {
199                        file_path: Some(file_path.clone()),
200                        error_type,
201                        message: format!("Failed to parse file: {}", e),
202                    });
203                    // Check graceful_degradation after parse failure
204                    if !self.config.graceful_degradation {
205                        return Ok(self.build_result(0, 0));
206                    }
207                }
208            }
209        }
210
211        // Early return if parsing failed and strict mode is enabled
212        if !self.config.graceful_degradation && !self.errors.is_empty() {
213            return Ok(self.build_result(0, 0));
214        }
215
216        // Step 3: Two-pass loading - UDTs first, then tables
217        let (udts_loaded, tables_loaded) = self.apply_schemas(parsed_schemas).await;
218
219        Ok(self.build_result(tables_loaded, udts_loaded))
220    }
221
222    /// Discover files from a path (file or directory)
223    fn discover_files(&mut self, path: &Path, files: &mut Vec<PathBuf>) -> Result<()> {
224        if !path.exists() {
225            return Err(Error::InvalidPath(format!(
226                "Path does not exist: {}",
227                path.display()
228            )));
229        }
230
231        if path.is_file() {
232            // Single file - validate extension
233            if let Some(ext) = path.extension() {
234                let ext_str = ext.to_string_lossy().to_lowercase();
235                if ext_str == "cql" || ext_str == "json" {
236                    files.push(path.to_path_buf());
237                } else {
238                    self.warnings.push(SchemaLoadWarning {
239                        file_path: Some(path.to_path_buf()),
240                        message: format!("Skipping file with unsupported extension: {}", ext_str),
241                    });
242                }
243            }
244        } else if path.is_dir() {
245            // Directory - scan recursively in lexical order
246            self.scan_directory_recursive(path, files)?;
247        }
248
249        Ok(())
250    }
251
252    /// Recursively scan directory for schema files in lexical order
253    #[allow(clippy::only_used_in_recursion)]
254    fn scan_directory_recursive(&mut self, dir: &Path, files: &mut Vec<PathBuf>) -> Result<()> {
255        let mut entries: Vec<PathBuf> = std::fs::read_dir(dir)
256            .map_err(Error::Io)?
257            .filter_map(|entry| entry.ok().map(|e| e.path()))
258            .collect();
259
260        // Sort lexically for deterministic ordering
261        entries.sort();
262
263        for entry in entries {
264            if entry.is_file() {
265                if let Some(ext) = entry.extension() {
266                    let ext_str = ext.to_string_lossy().to_lowercase();
267                    if ext_str == "cql" || ext_str == "json" {
268                        files.push(entry);
269                    }
270                }
271            } else if entry.is_dir() {
272                self.scan_directory_recursive(&entry, files)?;
273            }
274        }
275
276        Ok(())
277    }
278
279    /// Parse a single file into intermediate schema format
280    async fn parse_file(&self, path: &Path) -> Result<Option<ParsedSchema>> {
281        let ext = path
282            .extension()
283            .ok_or_else(|| Error::InvalidPath("File has no extension".to_string()))?;
284
285        let ext_str = ext.to_string_lossy().to_lowercase();
286
287        match ext_str.as_str() {
288            "cql" => self.parse_cql_file(path).await,
289            "json" => self.parse_json_file(path).await,
290            _ => Err(Error::InvalidPath(format!(
291                "Unsupported file extension: {}",
292                ext_str
293            ))),
294        }
295    }
296
297    /// Apply parsed schemas to registries (two-pass: UDTs first, then tables)
298    async fn apply_schemas(&mut self, parsed_schemas: Vec<ParsedSchema>) -> (usize, usize) {
299        // Pass 1: Register all UDTs with last-wins strategy
300        let mut udt_map: HashMap<String, (String, UdtTypeDef)> = HashMap::new(); // key: keyspace.udt_name -> (keyspace, UdtTypeDef)
301
302        for parsed in &parsed_schemas {
303            for (qualified_name, udt_def) in &parsed.udts {
304                // qualified_name is already "keyspace.typename" from parse_cql_file
305                udt_map.insert(
306                    qualified_name.clone(),
307                    (udt_def.keyspace.clone(), udt_def.clone()),
308                );
309            }
310        }
311
312        // Register UDTs in registry
313        let mut udts_loaded = 0;
314        {
315            let mut udt_registry = self.udt_registry.write().await;
316            for (_key, (_keyspace, udt_def)) in udt_map {
317                if self.config.validate_udt_dependencies {
318                    // Validate dependencies exist
319                    if let Err(e) = udt_registry.register_udt_with_validation(udt_def.clone()) {
320                        self.errors.push(SchemaLoadError {
321                            file_path: None,
322                            error_type: LoadErrorType::CircularUdtDependency,
323                            message: format!("UDT validation failed: {}", e),
324                        });
325                        // Check graceful_degradation after UDT validation failure
326                        if !self.config.graceful_degradation {
327                            // Return early with UDTs loaded so far, skip tables
328                            return (udts_loaded, 0);
329                        }
330                        continue;
331                    }
332                } else {
333                    udt_registry.register_udt(udt_def);
334                }
335                udts_loaded += 1;
336            }
337        }
338
339        // Early return after UDT phase if strict mode and errors exist
340        if !self.config.graceful_degradation && !self.errors.is_empty() {
341            return (udts_loaded, 0);
342        }
343
344        // Pass 2: Register all tables with last-wins strategy
345        let mut table_map: HashMap<String, TableSchema> = HashMap::new();
346
347        for parsed in &parsed_schemas {
348            for (qualified_name, table_schema) in &parsed.tables {
349                // qualified_name is already "keyspace.table" from parse_cql_file
350                table_map.insert(qualified_name.clone(), table_schema.clone());
351            }
352        }
353
354        // Register tables in registry
355        let mut tables_loaded = 0;
356        {
357            let registry = self.registry.write().await;
358            for (_key, table_schema) in table_map {
359                // Fail fast on undefined UDT references (issue #761): a column
360                // referencing a UDT not in the registry surfaces here with the
361                // missing type named, rather than later as a confusing
362                // parse/deserialization error. Gated by the same flag that
363                // controls UDT dependency validation.
364                if self.config.validate_udt_dependencies {
365                    let udt_registry = self.udt_registry.read().await;
366                    if let Err(e) = table_schema.validate_udt_references(&udt_registry) {
367                        self.errors.push(SchemaLoadError {
368                            file_path: None,
369                            error_type: LoadErrorType::ValidationFailed,
370                            message: format!(
371                                "Failed to register table '{}.{}': {}",
372                                table_schema.keyspace, table_schema.table, e
373                            ),
374                        });
375                        if !self.config.graceful_degradation {
376                            return (udts_loaded, tables_loaded);
377                        }
378                        continue;
379                    }
380                }
381                match registry
382                    .register_schema(
383                        table_schema.clone(),
384                        crate::schema::registry::SchemaSource::Manual,
385                    )
386                    .await
387                {
388                    Ok(_) => tables_loaded += 1,
389                    Err(e) => {
390                        self.errors.push(SchemaLoadError {
391                            file_path: None,
392                            error_type: LoadErrorType::ValidationFailed,
393                            message: format!(
394                                "Failed to register table '{}.{}': {}",
395                                table_schema.keyspace, table_schema.table, e
396                            ),
397                        });
398                        // Check graceful_degradation after table registration failure
399                        if !self.config.graceful_degradation {
400                            // Return early with counts so far
401                            return (udts_loaded, tables_loaded);
402                        }
403                    }
404                }
405            }
406        }
407
408        (udts_loaded, tables_loaded)
409    }
410
411    /// Build load result from current state
412    fn build_result(&self, schemas_loaded: usize, udts_loaded: usize) -> LoadResult {
413        LoadResult {
414            schemas_loaded,
415            udts_loaded,
416            errors: self.errors.clone(),
417            warnings: self.warnings.clone(),
418        }
419    }
420}
421
422/// Shared test fixtures for aggregator submodules.
423#[cfg(test)]
424pub(super) mod test_support {
425    use super::*;
426    use crate::platform::Platform;
427    use crate::schema::registry::{SchemaRegistry, SchemaRegistryConfig};
428    use crate::Config;
429    use std::io::Write;
430    pub(crate) use tempfile::TempDir;
431
432    pub(crate) async fn setup_test_aggregator() -> (SchemaAggregator, TempDir) {
433        let temp_dir = TempDir::new().unwrap();
434        let config = Config::default();
435        let platform = Arc::new(Platform::new(&config).await.unwrap());
436
437        let registry_config = SchemaRegistryConfig::default();
438        let registry = Arc::new(RwLock::new(
439            SchemaRegistry::new(registry_config, platform, config)
440                .await
441                .unwrap(),
442        ));
443        let udt_registry = Arc::new(RwLock::new(UdtRegistry::new()));
444
445        let aggregator = SchemaAggregator::new(registry, udt_registry, AggregatorConfig::default());
446
447        (aggregator, temp_dir)
448    }
449
450    pub(crate) fn write_file(dir: &Path, name: &str, content: &str) -> PathBuf {
451        let path = dir.join(name);
452        let mut file = std::fs::File::create(&path).unwrap();
453        file.write_all(content.as_bytes()).unwrap();
454        path
455    }
456}
457
458#[cfg(test)]
459mod tests {
460    use super::test_support::*;
461    use super::*;
462
463    #[tokio::test]
464    async fn test_error_type_mapping_io_error() {
465        let (mut aggregator, _temp_dir) = setup_test_aggregator().await;
466
467        // Test with a non-existent file to trigger IO error
468        let non_existent_path = PathBuf::from("/nonexistent/path/schema.json");
469        let result = aggregator
470            .load_from_paths(std::slice::from_ref(&non_existent_path))
471            .await
472            .unwrap();
473
474        assert_eq!(result.schemas_loaded, 0);
475        assert_eq!(result.errors.len(), 1);
476        assert!(matches!(
477            result.errors[0].error_type,
478            LoadErrorType::FileRead
479        ));
480        assert!(result.errors[0]
481            .message
482            .contains("Failed to discover files"));
483    }
484
485    #[tokio::test]
486    async fn test_error_type_mapping_invalid_json() {
487        let (mut aggregator, temp_dir) = setup_test_aggregator().await;
488
489        // Test with malformed JSON
490        let invalid_json = r#"{"keyspace": "ks", "table": "broken", invalid}"#;
491        let path = write_file(temp_dir.path(), "invalid.json", invalid_json);
492        let result = aggregator.load_from_paths(&[path]).await.unwrap();
493
494        assert_eq!(result.schemas_loaded, 0);
495        assert_eq!(result.errors.len(), 1);
496        assert!(matches!(
497            result.errors[0].error_type,
498            LoadErrorType::InvalidJson
499        ));
500        assert!(result.errors[0].message.contains("Failed to parse file"));
501        assert!(result.errors[0].message.contains("Invalid JSON"));
502    }
503
504    #[tokio::test]
505    async fn test_error_type_mapping_invalid_cql() {
506        let (mut aggregator, temp_dir) = setup_test_aggregator().await;
507
508        // Test with invalid CQL syntax
509        let invalid_cql = r#"
510        CREATE INVALID SYNTAX HERE
511        id uuid PRIMARY KEY
512        "#;
513        let path = write_file(temp_dir.path(), "invalid.cql", invalid_cql);
514        let result = aggregator.load_from_paths(&[path]).await.unwrap();
515
516        assert_eq!(result.schemas_loaded, 0);
517        assert_eq!(result.errors.len(), 1);
518        assert!(matches!(
519            result.errors[0].error_type,
520            LoadErrorType::InvalidCql
521        ));
522        assert!(result.errors[0].message.contains("Failed to parse file"));
523    }
524
525    #[tokio::test]
526    async fn test_error_message_preservation() {
527        let (mut aggregator, temp_dir) = setup_test_aggregator().await;
528
529        // Test that original error messages are preserved
530        let invalid_json = r#"{"keyspace": "ks""#; // Missing closing brace
531        let path = write_file(temp_dir.path(), "broken.json", invalid_json);
532        let result = aggregator
533            .load_from_paths(std::slice::from_ref(&path))
534            .await
535            .unwrap();
536
537        assert_eq!(result.errors.len(), 1);
538        // Error message should contain both "Failed to parse file" and the original error
539        assert!(result.errors[0].message.contains("Failed to parse file"));
540        assert!(result.errors[0].message.contains("Invalid JSON"));
541        // File path should be preserved
542        assert_eq!(result.errors[0].file_path, Some(path));
543    }
544
545    #[tokio::test]
546    async fn test_multiple_error_types_in_batch() {
547        let (mut aggregator, temp_dir) = setup_test_aggregator().await;
548
549        // Create multiple files with different error types
550        let invalid_json = r#"{"invalid json"#;
551        let invalid_cql = r#"INVALID CQL SYNTAX"#;
552
553        let json_path = write_file(temp_dir.path(), "bad.json", invalid_json);
554        let cql_path = write_file(temp_dir.path(), "bad.cql", invalid_cql);
555
556        let result = aggregator
557            .load_from_paths(&[json_path, cql_path])
558            .await
559            .unwrap();
560
561        assert_eq!(result.schemas_loaded, 0);
562        assert_eq!(result.errors.len(), 2);
563
564        // Find the JSON and CQL errors
565        let json_error = result
566            .errors
567            .iter()
568            .find(|e| {
569                e.file_path
570                    .as_ref()
571                    .unwrap()
572                    .to_str()
573                    .unwrap()
574                    .ends_with(".json")
575            })
576            .unwrap();
577        let cql_error = result
578            .errors
579            .iter()
580            .find(|e| {
581                e.file_path
582                    .as_ref()
583                    .unwrap()
584                    .to_str()
585                    .unwrap()
586                    .ends_with(".cql")
587            })
588            .unwrap();
589
590        // Verify correct error types
591        assert!(matches!(json_error.error_type, LoadErrorType::InvalidJson));
592        assert!(matches!(cql_error.error_type, LoadErrorType::InvalidCql));
593    }
594
595    #[tokio::test]
596    #[cfg(unix)]
597    async fn test_file_read_error_from_parse_file() {
598        use std::fs;
599        use std::os::unix::fs::PermissionsExt;
600
601        let (mut aggregator, temp_dir) = setup_test_aggregator().await;
602
603        // Create a file and make it unreadable (Unix-only test)
604        let json_content =
605            r#"{"keyspace": "ks", "table": "test", "columns": [], "partition_keys": ["id"]}"#;
606        let path = write_file(temp_dir.path(), "unreadable.json", json_content);
607
608        // Make file unreadable
609        let mut perms = fs::metadata(&path).unwrap().permissions();
610        perms.set_mode(0o000);
611        fs::set_permissions(&path, perms).unwrap();
612
613        // Privileged users (e.g. uid 0 in containerized CI) bypass file
614        // permissions, so the read-error precondition cannot be created.
615        if fs::File::open(&path).is_ok() {
616            return;
617        }
618
619        let result = aggregator
620            .load_from_paths(std::slice::from_ref(&path))
621            .await
622            .unwrap();
623
624        // Restore permissions for cleanup
625        let mut perms = fs::metadata(&path).unwrap().permissions();
626        perms.set_mode(0o644);
627        let _ = fs::set_permissions(&path, perms);
628
629        // Should have an IO error
630        assert_eq!(result.schemas_loaded, 0);
631        assert_eq!(result.errors.len(), 1);
632        assert!(matches!(
633            result.errors[0].error_type,
634            LoadErrorType::FileRead
635        ));
636    }
637}
cqlite_core/schema/aggregator/mod.rs

cqlite_core/schema/aggregator/
mod.rs
