1167 lines
36 KiB
Rust
1167 lines
36 KiB
Rust
//! Schema management with inference, evolution tracking, and drift detection
|
|
//!
|
|
//! This module provides comprehensive schema management capabilities including
|
|
//! automatic schema inference, schema evolution tracking, drift detection,
|
|
//! and schema validation.
|
|
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::fmt;
|
|
|
|
use chrono::{DateTime, Utc};
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use crate::{DataRecord, DataValue, Result, ValidationError};
|
|
|
|
/// Schema manager for handling schema operations
|
|
#[derive(Debug, Clone)]
|
|
pub struct SchemaManager {
|
|
/// Current active schemas by name
|
|
pub schemas: HashMap<String, Schema>,
|
|
/// Schema evolution history
|
|
pub evolution_history: HashMap<String, Vec<SchemaVersion>>,
|
|
/// Configuration for schema operations
|
|
pub config: SchemaConfig,
|
|
/// Schema inference engine
|
|
pub inference_engine: SchemaInference,
|
|
}
|
|
|
|
/// Configuration for schema management
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaConfig {
|
|
/// Enable automatic schema evolution
|
|
pub auto_evolution: bool,
|
|
/// Strictness level for schema validation
|
|
pub validation_strictness: ValidationStrictness,
|
|
/// Minimum confidence for schema inference
|
|
pub inference_confidence_threshold: f64,
|
|
/// Enable drift detection
|
|
pub drift_detection_enabled: bool,
|
|
/// Sample size for schema inference
|
|
pub inference_sample_size: usize,
|
|
/// Maximum number of schema versions to keep
|
|
pub max_schema_versions: usize,
|
|
}
|
|
|
|
/// Schema validation strictness levels
|
|
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
|
|
pub enum ValidationStrictness {
|
|
/// Lenient validation (warnings only)
|
|
Lenient,
|
|
/// Standard validation (errors for major violations)
|
|
Standard,
|
|
/// Strict validation (errors for any violations)
|
|
Strict,
|
|
}
|
|
|
|
/// Data schema representation
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct Schema {
|
|
/// Schema name/identifier
|
|
pub name: String,
|
|
/// Schema version
|
|
pub version: String,
|
|
/// Schema description
|
|
pub description: String,
|
|
/// Field definitions
|
|
pub fields: HashMap<String, FieldDefinition>,
|
|
/// Schema-level constraints
|
|
pub constraints: Vec<SchemaConstraint>,
|
|
/// Schema metadata
|
|
pub metadata: SchemaMetadata,
|
|
/// JSON Schema representation (if available)
|
|
pub json_schema: Option<serde_json::Value>,
|
|
/// Creation timestamp
|
|
pub created_at: DateTime<Utc>,
|
|
/// Last modified timestamp
|
|
pub modified_at: DateTime<Utc>,
|
|
}
|
|
|
|
/// Field definition in a schema
|
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
|
pub struct FieldDefinition {
|
|
/// Field name
|
|
pub name: String,
|
|
/// Data type
|
|
pub data_type: DataType,
|
|
/// Whether field is required
|
|
pub required: bool,
|
|
/// Whether field allows null values
|
|
pub nullable: bool,
|
|
/// Field description
|
|
pub description: String,
|
|
/// Field constraints
|
|
pub constraints: Vec<FieldConstraint>,
|
|
/// Default value (if any)
|
|
pub default_value: Option<DataValue>,
|
|
/// Field metadata
|
|
pub metadata: HashMap<String, String>,
|
|
}
|
|
|
|
/// Supported data types
|
|
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
|
pub enum DataType {
|
|
/// Boolean type
|
|
Boolean,
|
|
/// Integer type
|
|
Integer,
|
|
/// Floating point type
|
|
Float,
|
|
/// String type
|
|
String,
|
|
/// Date/timestamp type
|
|
Timestamp,
|
|
/// Array type (with element type)
|
|
Array(Box<DataType>),
|
|
/// Object type
|
|
Object,
|
|
/// Union type (multiple possible types)
|
|
Union(Vec<DataType>),
|
|
/// Null type
|
|
Null,
|
|
/// Unknown/any type
|
|
Any,
|
|
}
|
|
|
|
/// Field-level constraints
|
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
|
pub enum FieldConstraint {
|
|
/// Minimum value constraint
|
|
MinValue(f64),
|
|
/// Maximum value constraint
|
|
MaxValue(f64),
|
|
/// Minimum length constraint (for strings/arrays)
|
|
MinLength(usize),
|
|
/// Maximum length constraint (for strings/arrays)
|
|
MaxLength(usize),
|
|
/// Pattern constraint (regex)
|
|
Pattern(String),
|
|
/// Enumeration constraint (allowed values)
|
|
Enum(Vec<String>),
|
|
/// Unique constraint
|
|
Unique,
|
|
/// Custom constraint with description
|
|
Custom(String),
|
|
}
|
|
|
|
/// Schema-level constraints
|
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
|
pub enum SchemaConstraint {
|
|
/// Primary key constraint
|
|
PrimaryKey(Vec<String>),
|
|
/// Foreign key constraint
|
|
ForeignKey {
|
|
fields: Vec<String>,
|
|
references: String,
|
|
},
|
|
/// Unique constraint across multiple fields
|
|
UniqueComposite(Vec<String>),
|
|
/// Check constraint with condition
|
|
Check(String),
|
|
/// Custom schema constraint
|
|
Custom(String),
|
|
}
|
|
|
|
/// Schema metadata
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaMetadata {
|
|
/// Schema author
|
|
pub author: String,
|
|
/// Schema tags
|
|
pub tags: Vec<String>,
|
|
/// Schema category
|
|
pub category: String,
|
|
/// Data source information
|
|
pub data_source: Option<String>,
|
|
/// Usage statistics
|
|
pub usage_stats: UsageStats,
|
|
/// Schema quality metrics
|
|
pub quality_metrics: SchemaQualityMetrics,
|
|
}
|
|
|
|
/// Usage statistics for a schema
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct UsageStats {
|
|
/// Number of times schema was used for validation
|
|
pub validation_count: usize,
|
|
/// Number of validation failures
|
|
pub failure_count: usize,
|
|
/// Last used timestamp
|
|
pub last_used: Option<DateTime<Utc>>,
|
|
/// Average validation time in milliseconds
|
|
pub avg_validation_time_ms: f64,
|
|
}
|
|
|
|
/// Schema quality metrics
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaQualityMetrics {
|
|
/// Schema completeness score (0-1)
|
|
pub completeness: f64,
|
|
/// Schema consistency score (0-1)
|
|
pub consistency: f64,
|
|
/// Schema clarity score (0-1)
|
|
pub clarity: f64,
|
|
/// Overall schema quality score (0-1)
|
|
pub overall_quality: f64,
|
|
}
|
|
|
|
/// Schema version information
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaVersion {
|
|
/// Version number
|
|
pub version: String,
|
|
/// Schema at this version
|
|
pub schema: Schema,
|
|
/// Changes made in this version
|
|
pub changes: Vec<SchemaChange>,
|
|
/// Version timestamp
|
|
pub timestamp: DateTime<Utc>,
|
|
/// Migration script (if needed)
|
|
pub migration_script: Option<String>,
|
|
}
|
|
|
|
/// Types of schema changes
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum SchemaChange {
|
|
/// Field was added
|
|
FieldAdded {
|
|
name: String,
|
|
definition: FieldDefinition,
|
|
},
|
|
/// Field was removed
|
|
FieldRemoved { name: String },
|
|
/// Field was modified
|
|
FieldModified {
|
|
name: String,
|
|
old_def: FieldDefinition,
|
|
new_def: FieldDefinition,
|
|
},
|
|
/// Field was renamed
|
|
FieldRenamed { old_name: String, new_name: String },
|
|
/// Constraint was added
|
|
ConstraintAdded(SchemaConstraint),
|
|
/// Constraint was removed
|
|
ConstraintRemoved(SchemaConstraint),
|
|
/// Schema metadata changed
|
|
MetadataChanged {
|
|
old: SchemaMetadata,
|
|
new: SchemaMetadata,
|
|
},
|
|
}
|
|
|
|
/// Schema inference engine
|
|
#[derive(Debug, Clone)]
|
|
pub struct SchemaInference {
|
|
/// Configuration for inference
|
|
pub config: InferenceConfig,
|
|
/// Field type inference statistics
|
|
pub field_stats: HashMap<String, FieldInferenceStats>,
|
|
}
|
|
|
|
/// Configuration for schema inference
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct InferenceConfig {
|
|
/// Minimum confidence threshold for type inference
|
|
pub confidence_threshold: f64,
|
|
/// Sample size for inference
|
|
pub sample_size: usize,
|
|
/// Enable nullable field detection
|
|
pub detect_nullable: bool,
|
|
/// Enable constraint inference
|
|
pub infer_constraints: bool,
|
|
/// Enable pattern detection for strings
|
|
pub detect_patterns: bool,
|
|
}
|
|
|
|
/// Statistics for field type inference
|
|
#[derive(Debug, Clone)]
|
|
pub struct FieldInferenceStats {
|
|
/// Field name
|
|
pub field_name: String,
|
|
/// Type frequency counts
|
|
pub type_counts: HashMap<DataType, usize>,
|
|
/// Total samples seen
|
|
pub total_samples: usize,
|
|
/// Null value count
|
|
pub null_count: usize,
|
|
/// Sample values for analysis
|
|
pub sample_values: Vec<DataValue>,
|
|
}
|
|
|
|
/// Schema evolution tracking
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaEvolution {
|
|
/// Schema name
|
|
pub schema_name: String,
|
|
/// Evolution timeline
|
|
pub timeline: Vec<EvolutionEvent>,
|
|
/// Evolution statistics
|
|
pub stats: EvolutionStats,
|
|
}
|
|
|
|
/// Schema evolution event
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct EvolutionEvent {
|
|
/// Event timestamp
|
|
pub timestamp: DateTime<Utc>,
|
|
/// Event type
|
|
pub event_type: EvolutionEventType,
|
|
/// Event description
|
|
pub description: String,
|
|
/// Impact level
|
|
pub impact_level: ImpactLevel,
|
|
/// Affected components
|
|
pub affected_components: Vec<String>,
|
|
}
|
|
|
|
/// Types of evolution events
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum EvolutionEventType {
|
|
/// Schema creation
|
|
Created,
|
|
/// Schema update
|
|
Updated,
|
|
/// Schema deprecated
|
|
Deprecated,
|
|
/// Schema deleted
|
|
Deleted,
|
|
/// Breaking change detected
|
|
BreakingChange,
|
|
/// Non-breaking change detected
|
|
NonBreakingChange,
|
|
/// Schema migration
|
|
Migration,
|
|
}
|
|
|
|
/// Impact level of schema changes
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
|
pub enum ImpactLevel {
|
|
/// No impact
|
|
None,
|
|
/// Low impact
|
|
Low,
|
|
/// Medium impact
|
|
Medium,
|
|
/// High impact
|
|
High,
|
|
/// Critical impact
|
|
Critical,
|
|
}
|
|
|
|
/// Evolution statistics
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct EvolutionStats {
|
|
/// Total number of changes
|
|
pub total_changes: usize,
|
|
/// Breaking changes count
|
|
pub breaking_changes: usize,
|
|
/// Non-breaking changes count
|
|
pub non_breaking_changes: usize,
|
|
/// Average time between changes (in days)
|
|
pub avg_change_interval_days: f64,
|
|
/// Schema stability score (0-1)
|
|
pub stability_score: f64,
|
|
}
|
|
|
|
/// Schema drift detection result
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaDrift {
|
|
/// Schema name
|
|
pub schema_name: String,
|
|
/// Drift detection timestamp
|
|
pub detected_at: DateTime<Utc>,
|
|
/// Drift severity
|
|
pub severity: DriftSeverity,
|
|
/// Detected drift types
|
|
pub drift_types: Vec<DriftType>,
|
|
/// Drift score (0-1, higher = more drift)
|
|
pub drift_score: f64,
|
|
/// Recommended actions
|
|
pub recommendations: Vec<String>,
|
|
}
|
|
|
|
/// Schema drift severity levels
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
|
pub enum DriftSeverity {
|
|
/// No significant drift
|
|
None,
|
|
/// Minor drift detected
|
|
Minor,
|
|
/// Moderate drift detected
|
|
Moderate,
|
|
/// Major drift detected
|
|
Major,
|
|
/// Critical drift detected
|
|
Critical,
|
|
}
|
|
|
|
/// Types of schema drift
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum DriftType {
|
|
/// New fields detected
|
|
NewFields(Vec<String>),
|
|
/// Fields removed
|
|
RemovedFields(Vec<String>),
|
|
/// Type changes detected
|
|
TypeChanges(HashMap<String, (DataType, DataType)>),
|
|
/// Constraint violations
|
|
ConstraintViolations(Vec<String>),
|
|
/// Pattern changes in string fields
|
|
PatternChanges(HashMap<String, (String, String)>),
|
|
/// Value distribution changes
|
|
DistributionChanges(HashMap<String, f64>),
|
|
}
|
|
|
|
/// Schema validation result
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaValidation {
|
|
/// Whether validation passed
|
|
pub is_valid: bool,
|
|
/// Validation errors
|
|
pub errors: Vec<SchemaValidationError>,
|
|
/// Validation warnings
|
|
pub warnings: Vec<SchemaValidationWarning>,
|
|
/// Validation metadata
|
|
pub metadata: ValidationMetadata,
|
|
}
|
|
|
|
/// Schema validation error
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaValidationError {
|
|
/// Error code
|
|
pub code: String,
|
|
/// Error message
|
|
pub message: String,
|
|
/// Field path where error occurred
|
|
pub field_path: String,
|
|
/// Expected value or constraint
|
|
pub expected: String,
|
|
/// Actual value found
|
|
pub actual: String,
|
|
/// Error severity
|
|
pub severity: ErrorSeverity,
|
|
}
|
|
|
|
/// Schema validation warning
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaValidationWarning {
|
|
/// Warning code
|
|
pub code: String,
|
|
/// Warning message
|
|
pub message: String,
|
|
/// Field path where warning occurred
|
|
pub field_path: String,
|
|
/// Recommendation
|
|
pub recommendation: String,
|
|
}
|
|
|
|
/// Error severity levels
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
|
pub enum ErrorSeverity {
|
|
/// Information level
|
|
Info,
|
|
/// Warning level
|
|
Warning,
|
|
/// Error level
|
|
Error,
|
|
/// Critical error level
|
|
Critical,
|
|
}
|
|
|
|
/// Validation metadata
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct ValidationMetadata {
|
|
/// Schema version used for validation
|
|
pub schema_version: String,
|
|
/// Validation timestamp
|
|
pub validated_at: DateTime<Utc>,
|
|
/// Validation duration in milliseconds
|
|
pub duration_ms: u64,
|
|
/// Number of records validated
|
|
pub records_validated: usize,
|
|
/// Validation engine version
|
|
pub engine_version: String,
|
|
}
|
|
|
|
/// Schema comparison result
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct SchemaDiff {
|
|
/// Source schema name
|
|
pub source_schema: String,
|
|
/// Target schema name
|
|
pub target_schema: String,
|
|
/// Detected differences
|
|
pub differences: Vec<SchemaDifference>,
|
|
/// Compatibility assessment
|
|
pub compatibility: CompatibilityLevel,
|
|
/// Migration complexity
|
|
pub migration_complexity: MigrationComplexity,
|
|
}
|
|
|
|
/// Individual schema difference
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum SchemaDifference {
|
|
/// Field added
|
|
FieldAdded { name: String },
|
|
/// Field removed
|
|
FieldRemoved { name: String },
|
|
/// Field type changed
|
|
FieldTypeChanged {
|
|
name: String,
|
|
from: DataType,
|
|
to: DataType,
|
|
},
|
|
/// Field constraint changed
|
|
FieldConstraintChanged { field: String, constraint: String },
|
|
/// Schema constraint added
|
|
SchemaConstraintAdded(SchemaConstraint),
|
|
/// Schema constraint removed
|
|
SchemaConstraintRemoved(SchemaConstraint),
|
|
}
|
|
|
|
/// Schema compatibility levels
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
|
pub enum CompatibilityLevel {
|
|
/// Fully compatible (no breaking changes)
|
|
FullyCompatible,
|
|
/// Backward compatible (new schema can read old data)
|
|
BackwardCompatible,
|
|
/// Forward compatible (old schema can read new data)
|
|
ForwardCompatible,
|
|
/// Incompatible (breaking changes)
|
|
Incompatible,
|
|
}
|
|
|
|
/// Migration complexity levels
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
|
pub enum MigrationComplexity {
|
|
/// No migration needed
|
|
None,
|
|
/// Simple migration (automatic)
|
|
Simple,
|
|
/// Moderate migration (some manual steps)
|
|
Moderate,
|
|
/// Complex migration (significant changes)
|
|
Complex,
|
|
/// Very complex migration (major restructuring)
|
|
VeryComplex,
|
|
}
|
|
|
|
impl Default for SchemaConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
auto_evolution: false,
|
|
validation_strictness: ValidationStrictness::Standard,
|
|
inference_confidence_threshold: 0.8,
|
|
drift_detection_enabled: true,
|
|
inference_sample_size: 1000,
|
|
max_schema_versions: 10,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Default for InferenceConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
confidence_threshold: 0.8,
|
|
sample_size: 1000,
|
|
detect_nullable: true,
|
|
infer_constraints: true,
|
|
detect_patterns: true,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl SchemaManager {
|
|
/// Create a new schema manager
|
|
pub fn new() -> Self {
|
|
Self {
|
|
schemas: HashMap::new(),
|
|
evolution_history: HashMap::new(),
|
|
config: SchemaConfig::default(),
|
|
inference_engine: SchemaInference::new(),
|
|
}
|
|
}
|
|
|
|
/// Create schema manager with custom configuration
|
|
pub fn with_config(config: SchemaConfig) -> Self {
|
|
Self {
|
|
schemas: HashMap::new(),
|
|
evolution_history: HashMap::new(),
|
|
inference_engine: SchemaInference::with_config(InferenceConfig {
|
|
confidence_threshold: config.inference_confidence_threshold,
|
|
sample_size: config.inference_sample_size,
|
|
..InferenceConfig::default()
|
|
}),
|
|
config,
|
|
}
|
|
}
|
|
|
|
/// Register a new schema
|
|
pub fn register_schema(&mut self, schema: Schema) -> Result<()> {
|
|
let schema_name = schema.name.clone();
|
|
|
|
// Create initial version
|
|
let version = SchemaVersion {
|
|
version: schema.version.clone(),
|
|
schema: schema.clone(),
|
|
changes: vec![],
|
|
timestamp: Utc::now(),
|
|
migration_script: None,
|
|
};
|
|
|
|
// Add to history
|
|
self.evolution_history
|
|
.entry(schema_name.clone())
|
|
.or_insert_with(Vec::new)
|
|
.push(version);
|
|
|
|
// Set as active schema
|
|
self.schemas.insert(schema_name, schema);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Infer schema from sample data
|
|
pub fn infer_schema(&mut self, name: String, records: &[DataRecord]) -> Result<Schema> {
|
|
self.inference_engine.infer_schema(name, records)
|
|
}
|
|
|
|
/// Validate a record against a schema
|
|
pub fn validate_record(&self, record: &DataRecord) -> Result<SchemaValidation> {
|
|
// For this implementation, we'll do basic validation
|
|
// In a full implementation, this would use the JSON Schema validator
|
|
|
|
let mut errors = Vec::new();
|
|
let warnings = Vec::new();
|
|
|
|
// Check if we have a schema (simplified - would normally determine schema from record)
|
|
if self.schemas.is_empty() {
|
|
errors.push(SchemaValidationError {
|
|
code: "NO_SCHEMA".to_string(),
|
|
message: "No schema available for validation".to_string(),
|
|
field_path: "".to_string(),
|
|
expected: "Schema definition".to_string(),
|
|
actual: "No schema".to_string(),
|
|
severity: ErrorSeverity::Warning,
|
|
});
|
|
}
|
|
|
|
let is_valid = errors
|
|
.iter()
|
|
.all(|e| e.severity != ErrorSeverity::Error && e.severity != ErrorSeverity::Critical);
|
|
|
|
Ok(SchemaValidation {
|
|
is_valid,
|
|
errors,
|
|
warnings,
|
|
metadata: ValidationMetadata {
|
|
schema_version: "1.0.0".to_string(),
|
|
validated_at: Utc::now(),
|
|
duration_ms: 1,
|
|
records_validated: 1,
|
|
engine_version: "1.0.0".to_string(),
|
|
},
|
|
})
|
|
}
|
|
|
|
/// Detect schema drift
|
|
pub fn detect_drift(
|
|
&self,
|
|
schema_name: &str,
|
|
recent_records: &[DataRecord],
|
|
) -> Result<Option<SchemaDrift>> {
|
|
if !self.config.drift_detection_enabled {
|
|
return Ok(None);
|
|
}
|
|
|
|
if let Some(schema) = self.schemas.get(schema_name) {
|
|
// Simplified drift detection
|
|
let mut drift_types = Vec::new();
|
|
let mut drift_score = 0.0;
|
|
|
|
// Check for new fields not in schema
|
|
let mut all_fields = HashSet::new();
|
|
for record in recent_records {
|
|
all_fields.extend(record.fields.keys().cloned());
|
|
}
|
|
|
|
let schema_fields: HashSet<String> = schema.fields.keys().cloned().collect();
|
|
let new_fields: Vec<String> = all_fields.difference(&schema_fields).cloned().collect();
|
|
|
|
if !new_fields.is_empty() {
|
|
drift_types.push(DriftType::NewFields(new_fields));
|
|
drift_score += 0.3;
|
|
}
|
|
|
|
let severity = match drift_score {
|
|
s if s > 0.8 => DriftSeverity::Critical,
|
|
s if s > 0.6 => DriftSeverity::Major,
|
|
s if s > 0.4 => DriftSeverity::Moderate,
|
|
s if s > 0.2 => DriftSeverity::Minor,
|
|
_ => DriftSeverity::None,
|
|
};
|
|
|
|
if severity != DriftSeverity::None {
|
|
return Ok(Some(SchemaDrift {
|
|
schema_name: schema_name.to_string(),
|
|
detected_at: Utc::now(),
|
|
severity,
|
|
drift_types,
|
|
drift_score,
|
|
recommendations: vec![
|
|
"Consider updating schema to include new fields".to_string(),
|
|
],
|
|
}));
|
|
}
|
|
}
|
|
|
|
Ok(None)
|
|
}
|
|
|
|
/// Compare two schemas
|
|
pub fn compare_schemas(&self, source_name: &str, target_name: &str) -> Result<SchemaDiff> {
|
|
let source_schema = self.schemas.get(source_name).ok_or_else(|| {
|
|
ValidationError::Schema(format!("Source schema '{}' not found", source_name))
|
|
})?;
|
|
|
|
let target_schema = self.schemas.get(target_name).ok_or_else(|| {
|
|
ValidationError::Schema(format!("Target schema '{}' not found", target_name))
|
|
})?;
|
|
|
|
let mut differences = Vec::new();
|
|
|
|
// Compare fields
|
|
for (field_name, source_field) in &source_schema.fields {
|
|
if let Some(target_field) = target_schema.fields.get(field_name) {
|
|
if source_field.data_type != target_field.data_type {
|
|
differences.push(SchemaDifference::FieldTypeChanged {
|
|
name: field_name.clone(),
|
|
from: source_field.data_type.clone(),
|
|
to: target_field.data_type.clone(),
|
|
});
|
|
}
|
|
} else {
|
|
differences.push(SchemaDifference::FieldRemoved {
|
|
name: field_name.clone(),
|
|
});
|
|
}
|
|
}
|
|
|
|
// Check for new fields
|
|
for field_name in target_schema.fields.keys() {
|
|
if !source_schema.fields.contains_key(field_name) {
|
|
differences.push(SchemaDifference::FieldAdded {
|
|
name: field_name.clone(),
|
|
});
|
|
}
|
|
}
|
|
|
|
// Determine compatibility
|
|
let compatibility = if differences.is_empty() {
|
|
CompatibilityLevel::FullyCompatible
|
|
} else if differences.iter().any(|d| {
|
|
matches!(
|
|
d,
|
|
SchemaDifference::FieldRemoved { .. } | SchemaDifference::FieldTypeChanged { .. }
|
|
)
|
|
}) {
|
|
CompatibilityLevel::Incompatible
|
|
} else {
|
|
CompatibilityLevel::BackwardCompatible
|
|
};
|
|
|
|
// Determine migration complexity
|
|
let migration_complexity = match differences.len() {
|
|
0 => MigrationComplexity::None,
|
|
1..=2 => MigrationComplexity::Simple,
|
|
3..=5 => MigrationComplexity::Moderate,
|
|
6..=10 => MigrationComplexity::Complex,
|
|
_ => MigrationComplexity::VeryComplex,
|
|
};
|
|
|
|
Ok(SchemaDiff {
|
|
source_schema: source_name.to_string(),
|
|
target_schema: target_name.to_string(),
|
|
differences,
|
|
compatibility,
|
|
migration_complexity,
|
|
})
|
|
}
|
|
|
|
/// Get schema evolution history
|
|
pub fn get_evolution_history(&self, schema_name: &str) -> Option<&Vec<SchemaVersion>> {
|
|
self.evolution_history.get(schema_name)
|
|
}
|
|
}
|
|
|
|
impl Default for SchemaManager {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl SchemaInference {
|
|
/// Create a new schema inference engine
|
|
pub fn new() -> Self {
|
|
Self {
|
|
config: InferenceConfig::default(),
|
|
field_stats: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
/// Create inference engine with custom configuration
|
|
pub fn with_config(config: InferenceConfig) -> Self {
|
|
Self {
|
|
config,
|
|
field_stats: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
/// Infer schema from sample records
|
|
pub fn infer_schema(&mut self, name: String, records: &[DataRecord]) -> Result<Schema> {
|
|
if records.is_empty() {
|
|
return Err(ValidationError::Schema(
|
|
"Cannot infer schema from empty dataset".to_string(),
|
|
));
|
|
}
|
|
|
|
// Collect field statistics
|
|
self.collect_field_stats(records);
|
|
|
|
// Infer field definitions
|
|
let mut fields = HashMap::new();
|
|
for (field_name, stats) in &self.field_stats {
|
|
let field_def = self.infer_field_definition(field_name, stats)?;
|
|
fields.insert(field_name.clone(), field_def);
|
|
}
|
|
|
|
// Create schema
|
|
Ok(Schema {
|
|
name,
|
|
version: "1.0.0".to_string(),
|
|
description: "Automatically inferred schema".to_string(),
|
|
fields,
|
|
constraints: Vec::new(),
|
|
metadata: SchemaMetadata {
|
|
author: "Schema Inference Engine".to_string(),
|
|
tags: vec!["auto-generated".to_string()],
|
|
category: "inferred".to_string(),
|
|
data_source: None,
|
|
usage_stats: UsageStats {
|
|
validation_count: 0,
|
|
failure_count: 0,
|
|
last_used: None,
|
|
avg_validation_time_ms: 0.0,
|
|
},
|
|
quality_metrics: SchemaQualityMetrics {
|
|
completeness: 0.8,
|
|
consistency: 0.9,
|
|
clarity: 0.7,
|
|
overall_quality: 0.8,
|
|
},
|
|
},
|
|
json_schema: None,
|
|
created_at: Utc::now(),
|
|
modified_at: Utc::now(),
|
|
})
|
|
}
|
|
|
|
fn collect_field_stats(&mut self, records: &[DataRecord]) {
|
|
self.field_stats.clear();
|
|
|
|
let sample_limit = self.config.sample_size.min(records.len());
|
|
|
|
for record in records.iter().take(sample_limit) {
|
|
for (field_name, value) in &record.fields {
|
|
let stats = self
|
|
.field_stats
|
|
.entry(field_name.clone())
|
|
.or_insert_with(|| FieldInferenceStats {
|
|
field_name: field_name.clone(),
|
|
type_counts: HashMap::new(),
|
|
total_samples: 0,
|
|
null_count: 0,
|
|
sample_values: Vec::new(),
|
|
});
|
|
|
|
stats.total_samples += 1;
|
|
|
|
if value.is_null() {
|
|
stats.null_count += 1;
|
|
} else {
|
|
// Keep sample values for analysis (limit to 100)
|
|
if stats.sample_values.len() < 100 {
|
|
stats.sample_values.push(value.clone());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Calculate type counts in a separate pass to avoid borrow checker issues
|
|
for record in records.iter().take(sample_limit) {
|
|
for (field_name, value) in &record.fields {
|
|
if !value.is_null() {
|
|
let data_type = self.value_to_data_type(value);
|
|
if let Some(stats) = self.field_stats.get_mut(field_name) {
|
|
*stats.type_counts.entry(data_type).or_insert(0) += 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn infer_field_definition(
|
|
&self,
|
|
field_name: &str,
|
|
stats: &FieldInferenceStats,
|
|
) -> Result<FieldDefinition> {
|
|
// Determine the most likely data type
|
|
let data_type = if let Some((most_common_type, count)) =
|
|
stats.type_counts.iter().max_by_key(|(_, count)| **count)
|
|
{
|
|
let confidence = *count as f64 / stats.total_samples as f64;
|
|
if confidence >= self.config.confidence_threshold {
|
|
most_common_type.clone()
|
|
} else {
|
|
DataType::Any // Mixed types, use Any
|
|
}
|
|
} else {
|
|
DataType::Any
|
|
};
|
|
|
|
// Determine if field is nullable
|
|
let nullable = if self.config.detect_nullable {
|
|
stats.null_count > 0
|
|
} else {
|
|
false
|
|
};
|
|
|
|
// Determine if field is required (simplified logic)
|
|
let required = stats.null_count == 0 && stats.total_samples > 0;
|
|
|
|
// Infer constraints if enabled
|
|
let mut constraints = Vec::new();
|
|
if self.config.infer_constraints {
|
|
constraints.extend(self.infer_field_constraints(&data_type, &stats.sample_values));
|
|
}
|
|
|
|
Ok(FieldDefinition {
|
|
name: field_name.to_string(),
|
|
data_type,
|
|
required,
|
|
nullable,
|
|
description: format!(
|
|
"Inferred field (confidence: {:.2})",
|
|
*stats.type_counts.values().max().unwrap_or(&0) as f64 / stats.total_samples as f64
|
|
),
|
|
constraints,
|
|
default_value: None,
|
|
metadata: HashMap::new(),
|
|
})
|
|
}
|
|
|
|
fn value_to_data_type(&self, value: &DataValue) -> DataType {
|
|
match value {
|
|
DataValue::Null => DataType::Null,
|
|
DataValue::Bool(_) => DataType::Boolean,
|
|
DataValue::Int(_) => DataType::Integer,
|
|
DataValue::Float(_) => DataType::Float,
|
|
DataValue::String(_) => DataType::String,
|
|
DataValue::Array(_) => DataType::Array(Box::new(DataType::Any)),
|
|
DataValue::Object(_) => DataType::Object,
|
|
DataValue::Timestamp(_) => DataType::Timestamp,
|
|
}
|
|
}
|
|
|
|
fn infer_field_constraints(
|
|
&self,
|
|
data_type: &DataType,
|
|
sample_values: &[DataValue],
|
|
) -> Vec<FieldConstraint> {
|
|
let mut constraints = Vec::new();
|
|
|
|
match data_type {
|
|
DataType::String => {
|
|
if let Some(lengths) = self.collect_string_lengths(sample_values) {
|
|
if let (Some(&min_len), Some(&max_len)) =
|
|
(lengths.iter().min(), lengths.iter().max())
|
|
{
|
|
if min_len == max_len && min_len > 0 {
|
|
constraints.push(FieldConstraint::MinLength(min_len));
|
|
constraints.push(FieldConstraint::MaxLength(max_len));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
DataType::Integer | DataType::Float => {
|
|
if let Some(numbers) = self.collect_numeric_values(sample_values) {
|
|
if let (Some(min_val), Some(max_val)) = (
|
|
numbers.iter().min_by(|a, b| a.total_cmp(b)),
|
|
numbers.iter().max_by(|a, b| a.total_cmp(b)),
|
|
) {
|
|
// Only add constraints if there's a reasonable range
|
|
if (max_val - min_val) / min_val.abs() > 0.1 {
|
|
constraints.push(FieldConstraint::MinValue(*min_val));
|
|
constraints.push(FieldConstraint::MaxValue(*max_val));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
_ => {} // No constraints for other types in this simple implementation
|
|
}
|
|
|
|
constraints
|
|
}
|
|
|
|
fn collect_string_lengths(&self, values: &[DataValue]) -> Option<Vec<usize>> {
|
|
let lengths: Vec<usize> = values
|
|
.iter()
|
|
.filter_map(|v| v.as_string().map(|s| s.len()))
|
|
.collect();
|
|
|
|
if lengths.is_empty() {
|
|
None
|
|
} else {
|
|
Some(lengths)
|
|
}
|
|
}
|
|
|
|
fn collect_numeric_values(&self, values: &[DataValue]) -> Option<Vec<f64>> {
|
|
let numbers: Vec<f64> = values.iter().filter_map(|v| v.as_f64()).collect();
|
|
|
|
if numbers.is_empty() {
|
|
None
|
|
} else {
|
|
Some(numbers)
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Default for SchemaInference {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl fmt::Display for DataType {
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
match self {
|
|
DataType::Boolean => write!(f, "boolean"),
|
|
DataType::Integer => write!(f, "integer"),
|
|
DataType::Float => write!(f, "float"),
|
|
DataType::String => write!(f, "string"),
|
|
DataType::Timestamp => write!(f, "timestamp"),
|
|
DataType::Array(inner) => write!(f, "array[{}]", inner),
|
|
DataType::Object => write!(f, "object"),
|
|
DataType::Union(types) => {
|
|
let type_names: Vec<String> = types.iter().map(|t| t.to_string()).collect();
|
|
write!(f, "union[{}]", type_names.join("|"))
|
|
}
|
|
DataType::Null => write!(f, "null"),
|
|
DataType::Any => write!(f, "any"),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::DataValue;
|
|
use std::collections::HashMap;
|
|
|
|
#[test]
|
|
fn test_schema_manager_creation() {
|
|
let manager = SchemaManager::new();
|
|
assert!(manager.schemas.is_empty());
|
|
assert!(manager.evolution_history.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn test_schema_inference() {
|
|
let mut inference = SchemaInference::new();
|
|
|
|
let mut records = Vec::new();
|
|
for i in 0..5 {
|
|
let mut fields = HashMap::new();
|
|
fields.insert("id".to_string(), DataValue::Int(i));
|
|
fields.insert("name".to_string(), DataValue::String(format!("Name {}", i)));
|
|
fields.insert("active".to_string(), DataValue::Bool(i % 2 == 0));
|
|
|
|
records.push(DataRecord {
|
|
id: i.to_string(),
|
|
timestamp: Utc::now(),
|
|
fields,
|
|
metadata: HashMap::new(),
|
|
});
|
|
}
|
|
|
|
let schema = inference
|
|
.infer_schema("test_schema".to_string(), &records)
|
|
.unwrap();
|
|
|
|
assert_eq!(schema.name, "test_schema");
|
|
assert_eq!(schema.fields.len(), 3);
|
|
assert!(schema.fields.contains_key("id"));
|
|
assert!(schema.fields.contains_key("name"));
|
|
assert!(schema.fields.contains_key("active"));
|
|
}
|
|
|
|
#[test]
|
|
fn test_data_type_inference() {
|
|
let inference = SchemaInference::new();
|
|
|
|
assert_eq!(
|
|
inference.value_to_data_type(&DataValue::Bool(true)),
|
|
DataType::Boolean
|
|
);
|
|
assert_eq!(
|
|
inference.value_to_data_type(&DataValue::Int(42)),
|
|
DataType::Integer
|
|
);
|
|
assert_eq!(
|
|
inference.value_to_data_type(&DataValue::Float(3.14)),
|
|
DataType::Float
|
|
);
|
|
assert_eq!(
|
|
inference.value_to_data_type(&DataValue::String("test".to_string())),
|
|
DataType::String
|
|
);
|
|
assert_eq!(
|
|
inference.value_to_data_type(&DataValue::Null),
|
|
DataType::Null
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_schema_validation() {
|
|
let manager = SchemaManager::new();
|
|
|
|
let record = DataRecord {
|
|
id: "test".to_string(),
|
|
timestamp: Utc::now(),
|
|
fields: HashMap::new(),
|
|
metadata: HashMap::new(),
|
|
};
|
|
|
|
let validation = manager.validate_record(&record).unwrap();
|
|
// Should have warnings about no schema
|
|
assert!(!validation.warnings.is_empty() || !validation.errors.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn test_impact_level_ordering() {
|
|
assert!(ImpactLevel::None < ImpactLevel::Low);
|
|
assert!(ImpactLevel::Low < ImpactLevel::Medium);
|
|
assert!(ImpactLevel::Medium < ImpactLevel::High);
|
|
assert!(ImpactLevel::High < ImpactLevel::Critical);
|
|
}
|
|
|
|
#[test]
|
|
fn test_migration_complexity_ordering() {
|
|
assert!(MigrationComplexity::None < MigrationComplexity::Simple);
|
|
assert!(MigrationComplexity::Simple < MigrationComplexity::Moderate);
|
|
assert!(MigrationComplexity::Moderate < MigrationComplexity::Complex);
|
|
assert!(MigrationComplexity::Complex < MigrationComplexity::VeryComplex);
|
|
}
|
|
}
|