## Coverage Report

#### Created: 2025-09-08 21:26

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /home/noah/src/ruchy/src/runtime/dataflow_debugger.rs

    Line
    Count
    Source
    1
    //! Dataflow debugger for `DataFrame` pipeline debugging (RUCHY-0818)
    2
    //!
    3
    //! Provides comprehensive debugging capabilities for `DataFrame` operations,
    4
    //! including breakpoints, materialization on demand, and stage-by-stage analysis.
    5
    6
    use anyhow::Result;
    7
    use serde::{Deserialize, Serialize};
    8
    use std::collections::{HashMap, VecDeque};
    9
    use std::fmt;
    10
    use std::sync::{Arc, Mutex};
    11
    use std::time::{Duration, Instant};
    12
    13
    /// Dataflow debugger for `DataFrame` pipeline analysis
    14
    pub struct DataflowDebugger {
    15
        /// Pipeline execution stages
    16
        #[allow(dead_code)] // Future feature for pipeline management
    17
        pipeline_stages: Arc<Mutex<Vec<PipelineStage>>>,
    18
    19
        /// Active breakpoints by stage name
    20
        breakpoints: Arc<Mutex<HashMap<String, Breakpoint>>>,
    21
    22
        /// Materialized data at breakpoints
    23
        materialized_data: Arc<Mutex<HashMap<String, MaterializedFrame>>>,
    24
    25
        /// Debugger configuration
    26
        config: DataflowConfig,
    27
    28
        /// Execution history for analysis
    29
        execution_history: Arc<Mutex<VecDeque<ExecutionEvent>>>,
    30
    31
        /// Performance metrics per stage
    32
        stage_metrics: Arc<Mutex<HashMap<String, StageMetrics>>>,
    33
    34
        /// Current debugging session state
    35
        session_state: Arc<Mutex<SessionState>>,
    36
    }
    37
    38
    /// Configuration for the dataflow debugger
    39
    #[derive(Debug, Clone, Serialize, Deserialize)]
    40
    pub struct DataflowConfig {
    41
        /// Maximum number of rows to materialize at each stage
    42
        pub max_rows_per_stage: usize,
    43
    44
        /// Enable automatic materialization at each stage
    45
        pub auto_materialize: bool,
    46
    47
        /// Maximum execution history events to keep
    48
        pub max_history_events: usize,
    49
    50
        /// Enable performance profiling
    51
        pub enable_profiling: bool,
    52
    53
        /// Timeout for stage execution (in milliseconds)
    54
        pub stage_timeout_ms: u64,
    55
    56
        /// Enable detailed memory tracking
    57
        pub track_memory: bool,
    58
    59
        /// Enable diff computation between stages
    60
        pub compute_diffs: bool,
    61
    62
        /// Sample rate for large datasets (0.0-1.0)
    63
        pub sample_rate: f64,
    64
    }
    65
    66
    /// Individual stage in the `DataFrame` pipeline
    67
    #[derive(Debug, Clone, Serialize, Deserialize)]
    68
    pub struct PipelineStage {
    69
        /// Unique identifier for the stage
    70
        pub stage_id: String,
    71
    72
        /// Human-readable name
    73
        pub stage_name: String,
    74
    75
        /// Stage type (filter, map, `group_by`, etc.)
    76
        pub stage_type: StageType,
    77
    78
        /// Stage execution status
    79
        pub status: StageStatus,
    80
    81
        /// Input `DataFrame` schema
    82
        pub input_schema: Option<DataSchema>,
    83
    84
        /// Output `DataFrame` schema
    85
        pub output_schema: Option<DataSchema>,
    86
    87
        /// Stage execution time
    88
        pub execution_time: Option<Duration>,
    89
    90
        /// Memory usage for this stage
    91
        pub memory_usage: Option<usize>,
    92
    93
        /// Number of rows processed
    94
        pub rows_processed: Option<usize>,
    95
    96
        /// Stage-specific metadata
    97
        pub metadata: HashMap<String, String>,
    98
    }
    99
    100
    /// Types of `DataFrame` operations
    101
    #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
    102
    pub enum StageType {
    103
        /// Data loading stage
    104
        Load,
    105
        /// Filtering operations
    106
        Filter,
    107
        /// Column selection/projection
    108
        Select,
    109
        /// Column transformations
    110
        Map,
    111
        /// Grouping operations
    112
        GroupBy,
    113
        /// Aggregation operations
    114
        Aggregate,
    115
        /// Join operations
    116
        Join,
    117
        /// Sorting operations
    118
        Sort,
    119
        /// Window functions
    120
        Window,
    121
        /// Union operations
    122
        Union,
    123
        /// Custom user-defined operations
    124
        Custom(String),
    125
    }
    126
    127
    /// Execution status of a pipeline stage
    128
    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
    129
    pub enum StageStatus {
    130
        /// Stage not yet executed
    131
        Pending,
    132
        /// Stage currently executing
    133
        Running,
    134
        /// Stage completed successfully
    135
        Completed,
    136
        /// Stage failed with error
    137
        Failed(String),
    138
        /// Stage execution was cancelled
    139
        Cancelled,
    140
        /// Stage paused at breakpoint
    141
        Paused,
    142
    }
    143
    144
    /// Breakpoint configuration for pipeline debugging
    145
    #[derive(Debug, Clone, Serialize, Deserialize)]
    146
    pub struct Breakpoint {
    147
        /// Stage ID where breakpoint is set
    148
        pub stage_id: String,
    149
    150
        /// Breakpoint condition (optional)
    151
        pub condition: Option<BreakpointCondition>,
    152
    153
        /// Whether breakpoint is currently active
    154
        pub active: bool,
    155
    156
        /// Hit count for this breakpoint
    157
        pub hit_count: usize,
    158
    159
        /// Actions to perform when breakpoint is hit
    160
        pub actions: Vec<BreakpointAction>,
    161
    }
    162
    163
    /// Conditions for triggering breakpoints
    164
    #[derive(Debug, Clone, Serialize, Deserialize)]
    165
    pub enum BreakpointCondition {
    166
        /// Always break at this stage
    167
        Always,
    168
        /// Break if row count meets criteria
    169
        RowCount { operator: ComparisonOp, value: usize },
    170
        /// Break if execution time exceeds threshold
    171
        ExecutionTime { threshold_ms: u64 },
    172
        /// Break if memory usage exceeds threshold
    173
        MemoryUsage { threshold_mb: usize },
    174
        /// Break on specific data values
    175
        DataValue { column: String, value: DataValue },
    176
        /// Break on error conditions
    177
        OnError,
    178
    }
    179
    180
    /// Comparison operators for breakpoint conditions
    181
    #[derive(Debug, Clone, Serialize, Deserialize)]
    182
    pub enum ComparisonOp {
    183
        Equal,
    184
        NotEqual,
    185
        GreaterThan,
    186
        GreaterThanOrEqual,
    187
        LessThan,
    188
        LessThanOrEqual,
    189
    }
    190
    191
    /// Actions to perform when breakpoint is triggered
    192
    #[derive(Debug, Clone, Serialize, Deserialize)]
    193
    pub enum BreakpointAction {
    194
        /// Pause execution and wait for user input
    195
        Pause,
    196
        /// Print debug information
    197
        Print(String),
    198
        /// Materialize current `DataFrame`
    199
        Materialize,
    200
        /// Compute diff with previous stage
    201
        ComputeDiff,
    202
        /// Export data to file
    203
        Export { format: ExportFormat, path: String },
    204
    }
    205
    206
    /// Materialized `DataFrame` data for inspection
    207
    #[derive(Debug, Clone, Serialize, Deserialize)]
    208
    pub struct MaterializedFrame {
    209
        /// Stage ID where data was materialized
    210
        pub stage_id: String,
    211
    212
        /// `DataFrame` schema
    213
        pub schema: DataSchema,
    214
    215
        /// Sample of data rows (limited by config)
    216
        pub sample_data: Vec<DataRow>,
    217
    218
        /// Total number of rows in full dataset
    219
        pub total_rows: usize,
    220
    221
        /// Materialization timestamp
    222
        pub timestamp: std::time::SystemTime,
    223
    224
        /// Memory footprint of materialized data
    225
        pub memory_size: usize,
    226
    }
    227
    228
    /// `DataFrame` schema information
    229
    #[derive(Debug, Clone, Serialize, Deserialize)]
    230
    pub struct DataSchema {
    231
        /// Column definitions
    232
        pub columns: Vec<ColumnDef>,
    233
    234
        /// Schema hash for change detection
    235
        pub schema_hash: u64,
    236
    }
    237
    238
    /// Column definition in `DataFrame` schema
    239
    #[derive(Debug, Clone, Serialize, Deserialize)]
    240
    pub struct ColumnDef {
    241
        /// Column name
    242
        pub name: String,
    243
    244
        /// Data type
    245
        pub data_type: DataType,
    246
    247
        /// Whether column allows null values
    248
        pub nullable: bool,
    249
    }
    250
    251
    /// Supported data types in `DataFrames`
    252
    #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
    253
    pub enum DataType {
    254
        Boolean,
    255
        Integer,
    256
        Float,
    257
        String,
    258
        DateTime,
    259
        Array(Box<DataType>),
    260
        Struct(Vec<(String, DataType)>),
    261
    }
    262
    263
    /// Single row of data in materialized `DataFrame`
    264
    #[derive(Debug, Clone, Serialize, Deserialize)]
    265
    pub struct DataRow {
    266
        /// Values for each column
    267
        pub values: Vec<DataValue>,
    268
    }
    269
    270
    /// Individual data value in a `DataFrame` cell
    271
    #[derive(Debug, Clone, Serialize, Deserialize)]
    272
    pub enum DataValue {
    273
        Boolean(bool),
    274
        Integer(i64),
    275
        Float(f64),
    276
        String(String),
    277
        Null,
    278
        Array(Vec<DataValue>),
    279
    }
    280
    281
    /// Performance metrics for a pipeline stage
    282
    #[derive(Debug, Clone, Serialize, Deserialize)]
    283
    pub struct StageMetrics {
    284
        /// Stage execution time
    285
        pub execution_time: Duration,
    286
    287
        /// Memory peak usage during stage
    288
        pub peak_memory: usize,
    289
    290
        /// Number of rows input to stage
    291
        pub input_rows: usize,
    292
    293
        /// Number of rows output from stage
    294
        pub output_rows: usize,
    295
    296
        /// CPU time spent in stage
    297
        pub cpu_time: Duration,
    298
    299
        /// I/O operations performed
    300
        pub io_operations: usize,
    301
    302
        /// Cache hit ratio (if applicable)
    303
        pub cache_hit_ratio: Option<f64>,
    304
    }
    305
    306
    /// Execution event in the debugging session
    307
    #[derive(Debug, Clone, Serialize, Deserialize)]
    308
    pub struct ExecutionEvent {
    309
        /// Event timestamp
    310
        pub timestamp: std::time::SystemTime,
    311
    312
        /// Event type
    313
        pub event_type: EventType,
    314
    315
        /// Stage ID associated with event
    316
        pub stage_id: String,
    317
    318
        /// Additional event data
    319
        pub data: HashMap<String, String>,
    320
    }
    321
    322
    /// Types of execution events
    323
    #[derive(Debug, Clone, Serialize, Deserialize)]
    324
    pub enum EventType {
    325
        StageStarted,
    326
        StageCompleted,
    327
        StageFailed,
    328
        BreakpointHit,
    329
        DataMaterialized,
    330
        DiffComputed,
    331
        PerformanceAlert,
    332
    }
    333
    334
    /// Current state of the debugging session
    335
    #[derive(Debug, Clone, Serialize, Deserialize)]
    336
    pub struct SessionState {
    337
        /// Whether debugger is actively running
    338
        pub active: bool,
    339
    340
        /// Current stage being executed or paused at
    341
        pub current_stage: Option<String>,
    342
    343
        /// Session start time
    344
        pub session_start: std::time::SystemTime,
    345
    346
        /// Total execution time so far
    347
        pub total_execution_time: Duration,
    348
    349
        /// Number of breakpoints hit
    350
        pub breakpoints_hit: usize,
    351
    352
        /// Session metadata
    353
        pub metadata: HashMap<String, String>,
    354
    }
    355
    356
    /// Export formats for dataflow debugging data
    357
    #[derive(Debug, Clone, Serialize, Deserialize)]
    358
    pub enum ExportFormat {
    359
        /// Comma-separated values
    360
        Csv,
    361
        /// JSON format
    362
        Json,
    363
        /// Parquet format
    364
        Parquet,
    365
        /// Debug text format
    366
        Debug,
    367
    }
    368
    369
    impl Default for DataflowConfig {
    370
    0
        fn default() -> Self {
    371
    0
            Self {
    372
    0
                max_rows_per_stage: 1000,
    373
    0
                auto_materialize: false,
    374
    0
                max_history_events: 10000,
    375
    0
                enable_profiling: true,
    376
    0
                stage_timeout_ms: 30000, // 30 seconds
    377
    0
                track_memory: true,
    378
    0
                compute_diffs: false,
    379
    0
                sample_rate: 1.0, // No sampling by default
    380
    0
            }
    381
    0
        }
    382
    }
    383
    384
    impl DataflowDebugger {
    385
        /// Create a new dataflow debugger
    386
    0
        pub fn new(config: DataflowConfig) -> Self {
    387
    0
            Self {
    388
    0
                pipeline_stages: Arc::new(Mutex::new(Vec::new())),
    389
    0
                breakpoints: Arc::new(Mutex::new(HashMap::new())),
    390
    0
                materialized_data: Arc::new(Mutex::new(HashMap::new())),
    391
    0
                config,
    392
    0
                execution_history: Arc::new(Mutex::new(VecDeque::new())),
    393
    0
                stage_metrics: Arc::new(Mutex::new(HashMap::new())),
    394
    0
                session_state: Arc::new(Mutex::new(SessionState::default())),
    395
    0
            }
    396
    0
        }
    397
    398
        /// Start a new debugging session
    399
    0
        pub fn start_session(&self) -> Result<()> {
    400
    0
            let mut state = self.session_state
    401
    0
                .lock()
    402
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire session state lock"))?;
    403
    404
    0
            state.active = true;
    405
    0
            state.session_start = std::time::SystemTime::now();
    406
    0
            state.total_execution_time = Duration::from_secs(0);
    407
    0
            state.breakpoints_hit = 0;
    408
    0
            state.current_stage = None;
    409
    410
    0
            self.record_event(EventType::StageStarted, "session".to_string(), HashMap::new())?;
    411
    0
            Ok(())
    412
    0
        }
    413
    414
        /// Add a breakpoint to the debugger
    415
    0
        pub fn add_breakpoint(&self, breakpoint: Breakpoint) -> Result<()> {
    416
    0
            let mut breakpoints = self.breakpoints
    417
    0
                .lock()
    418
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire breakpoints lock"))?;
    419
    420
    0
            breakpoints.insert(breakpoint.stage_id.clone(), breakpoint);
    421
    0
            Ok(())
    422
    0
        }
    423
    424
        /// Remove a breakpoint by stage ID
    425
    0
        pub fn remove_breakpoint(&self, stage_id: &str) -> Result<bool> {
    426
    0
            let mut breakpoints = self.breakpoints
    427
    0
                .lock()
    428
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire breakpoints lock"))?;
    429
    430
    0
            Ok(breakpoints.remove(stage_id).is_some())
    431
    0
        }
    432
    433
        /// Execute a pipeline stage with debugging support
    434
    0
        pub fn execute_stage(&self, pipeline_stage: &mut PipelineStage) -> Result<StageExecutionResult> {
    435
    0
            let start_time = Instant::now();
    436
    0
            pipeline_stage.status = StageStatus::Running;
    437
    438
            // Update session state
    439
            {
    440
    0
                let mut state = self.session_state
    441
    0
                    .lock()
    442
    0
                    .map_err(|_| anyhow::anyhow!("Failed to acquire session state lock"))?;
    443
    0
                state.current_stage = Some(pipeline_stage.stage_id.clone());
    444
            }
    445
    446
            // Check for breakpoints
    447
    0
            if let Some(breakpoint) = self.check_breakpoint(&pipeline_stage.stage_id)? {
    448
    0
                if self.should_break(pipeline_stage, &breakpoint)? {
    449
    0
                    pipeline_stage.status = StageStatus::Paused;
    450
    0
                    self.handle_breakpoint_hit(&pipeline_stage.stage_id, &breakpoint)?;
    451
    0
                    return Ok(StageExecutionResult::Paused);
    452
    0
                }
    453
    0
            }
    454
    455
            // Simulate stage execution (in real implementation, this would execute actual DataFrame operations)
    456
    0
            std::thread::sleep(Duration::from_millis(10)); // Simulate work
    457
    458
            // Record execution metrics
    459
    0
            let execution_time = start_time.elapsed();
    460
    0
            pipeline_stage.execution_time = Some(execution_time);
    461
    0
            pipeline_stage.status = StageStatus::Completed;
    462
    463
            // Store performance metrics
    464
    0
            let metrics = StageMetrics {
    465
    0
                execution_time,
    466
    0
                peak_memory: 1024 * 1024, // 1MB simulated
    467
    0
                input_rows: pipeline_stage.rows_processed.unwrap_or(0),
    468
    0
                output_rows: pipeline_stage.rows_processed.unwrap_or(0),
    469
    0
                cpu_time: execution_time,
    470
    0
                io_operations: 1,
    471
    0
                cache_hit_ratio: Some(0.85),
    472
    0
            };
    473
    474
    0
            let mut stage_metrics = self.stage_metrics
    475
    0
                .lock()
    476
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire stage metrics lock"))?;
    477
    0
            stage_metrics.insert(pipeline_stage.stage_id.clone(), metrics);
    478
    479
            // Auto-materialize if configured
    480
    0
            if self.config.auto_materialize {
    481
    0
                self.materialize_stage(&pipeline_stage.stage_id)?;
    482
    0
            }
    483
    484
    0
            self.record_event(
    485
    0
                EventType::StageCompleted,
    486
    0
                pipeline_stage.stage_id.clone(),
    487
    0
                HashMap::from([("duration_ms".to_string(), execution_time.as_millis().to_string())])
    488
    0
            )?;
    489
    490
    0
            Ok(StageExecutionResult::Completed)
    491
    0
        }
    492
    493
        /// Materialize `DataFrame` data at a specific stage
    494
    0
        pub fn materialize_stage(&self, stage_id: &str) -> Result<MaterializedFrame> {
    495
            // In a real implementation, this would materialize actual DataFrame data
    496
    0
            let materialized = MaterializedFrame {
    497
    0
                stage_id: stage_id.to_string(),
    498
    0
                schema: DataSchema {
    499
    0
                    columns: vec![
    500
    0
                        ColumnDef {
    501
    0
                            name: "id".to_string(),
    502
    0
                            data_type: DataType::Integer,
    503
    0
                            nullable: false,
    504
    0
                        },
    505
    0
                        ColumnDef {
    506
    0
                            name: "name".to_string(),
    507
    0
                            data_type: DataType::String,
    508
    0
                            nullable: true,
    509
    0
                        },
    510
    0
                    ],
    511
    0
                    schema_hash: 12345,
    512
    0
                },
    513
    0
                sample_data: vec![
    514
    0
                    DataRow {
    515
    0
                        values: vec![DataValue::Integer(1), DataValue::String("Alice".to_string())],
    516
    0
                    },
    517
    0
                    DataRow {
    518
    0
                        values: vec![DataValue::Integer(2), DataValue::String("Bob".to_string())],
    519
    0
                    },
    520
    0
                ],
    521
    0
                total_rows: 1000,
    522
    0
                timestamp: std::time::SystemTime::now(),
    523
    0
                memory_size: 1024 * 50, // 50KB
    524
    0
            };
    525
    526
    0
            let mut materialized_data = self.materialized_data
    527
    0
                .lock()
    528
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire materialized data lock"))?;
    529
    530
    0
            materialized_data.insert(stage_id.to_string(), materialized.clone());
    531
    532
    0
            self.record_event(
    533
    0
                EventType::DataMaterialized,
    534
    0
                stage_id.to_string(),
    535
    0
                HashMap::from([("rows".to_string(), materialized.total_rows.to_string())])
    536
    0
            )?;
    537
    538
    0
            Ok(materialized)
    539
    0
        }
    540
    541
        /// Compute diff between two pipeline stages
    542
    0
        pub fn compute_stage_diff(&self, stage1_id: &str, stage2_id: &str) -> Result<StageDiff> {
    543
    0
            let materialized_data = self.materialized_data
    544
    0
                .lock()
    545
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire materialized data lock"))?;
    546
    547
    0
            let stage1_data = materialized_data.get(stage1_id)
    548
    0
                .ok_or_else(|| anyhow::anyhow!("Stage {} not materialized", stage1_id))?;
    549
    0
            let stage2_data = materialized_data.get(stage2_id)
    550
    0
                .ok_or_else(|| anyhow::anyhow!("Stage {} not materialized", stage2_id))?;
    551
    552
            // Compute basic diff metrics
    553
    0
            let row_count_diff = stage2_data.total_rows as i64 - stage1_data.total_rows as i64;
    554
    0
            let schema_changed = stage1_data.schema.schema_hash != stage2_data.schema.schema_hash;
    555
    556
    0
            let diff = StageDiff {
    557
    0
                stage1_id: stage1_id.to_string(),
    558
    0
                stage2_id: stage2_id.to_string(),
    559
    0
                row_count_diff,
    560
    0
                schema_changed,
    561
    0
                column_changes: self.compute_column_changes(&stage1_data.schema, &stage2_data.schema),
    562
    0
                data_changes: self.compute_data_changes(&stage1_data.sample_data, &stage2_data.sample_data),
    563
    0
            };
    564
    565
    0
            self.record_event(
    566
    0
                EventType::DiffComputed,
    567
    0
                format!("{stage1_id}:{stage2_id}"),
    568
    0
                HashMap::from([("row_diff".to_string(), row_count_diff.to_string())])
    569
    0
            )?;
    570
    571
    0
            Ok(diff)
    572
    0
        }
    573
    574
        /// Get current debugging session status
    575
    0
        pub fn get_session_status(&self) -> Result<SessionState> {
    576
    0
            let state = self.session_state
    577
    0
                .lock()
    578
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire session state lock"))?;
    579
    580
    0
            Ok(state.clone())
    581
    0
        }
    582
    583
        /// Get execution history
    584
    0
        pub fn get_execution_history(&self) -> Result<Vec<ExecutionEvent>> {
    585
    0
            let history = self.execution_history
    586
    0
                .lock()
    587
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire execution history lock"))?;
    588
    589
    0
            Ok(history.iter().cloned().collect())
    590
    0
        }
    591
    592
        /// Get performance metrics for all stages
    593
    0
        pub fn get_stage_metrics(&self) -> Result<HashMap<String, StageMetrics>> {
    594
    0
            let metrics = self.stage_metrics
    595
    0
                .lock()
    596
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire stage metrics lock"))?;
    597
    598
    0
            Ok(metrics.clone())
    599
    0
        }
    600
    601
        /// Export debugging data to various formats
    602
    0
        pub fn export_debug_data(&self, format: ExportFormat, output_path: &str) -> Result<()> {
    603
    0
            let history = self.get_execution_history()?;
    604
    0
            let metrics = self.get_stage_metrics()?;
    605
    0
            let session_status = self.get_session_status()?;
    606
    607
    0
            let debug_data = DebugExport {
    608
    0
                session_status,
    609
    0
                execution_history: history,
    610
    0
                stage_metrics: metrics,
    611
                materialized_data: {
    612
    0
                    let data = self.materialized_data
    613
    0
                        .lock()
    614
    0
                        .map_err(|_| anyhow::anyhow!("Failed to acquire materialized data lock"))?;
    615
    0
                    data.clone()
    616
                },
    617
            };
    618
    619
    0
            match format {
    620
                ExportFormat::Json => {
    621
    0
                    let json_data = serde_json::to_string_pretty(&debug_data)?;
    622
    0
                    std::fs::write(output_path, json_data)?;
    623
                }
    624
                ExportFormat::Debug => {
    625
    0
                    let debug_str = format!("{debug_data:#?}");
    626
    0
                    std::fs::write(output_path, debug_str)?;
    627
                }
    628
                _ => {
    629
    0
                    return Err(anyhow::anyhow!("Export format {:?} not yet implemented", format));
    630
                }
    631
            }
    632
    633
    0
            Ok(())
    634
    0
        }
    635
    636
        // Helper methods
    637
    638
    0
        fn check_breakpoint(&self, stage_id: &str) -> Result<Option<Breakpoint>> {
    639
    0
            let breakpoints = self.breakpoints
    640
    0
                .lock()
    641
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire breakpoints lock"))?;
    642
    643
    0
            Ok(breakpoints.get(stage_id).cloned())
    644
    0
        }
    645
    646
    0
        fn should_break(&self, _stage: &PipelineStage, breakpoint: &Breakpoint) -> Result<bool> {
    647
    0
            if !breakpoint.active {
    648
    0
                return Ok(false);
    649
    0
            }
    650
    651
    0
            match &breakpoint.condition {
    652
    0
                None | Some(BreakpointCondition::Always) => Ok(true),
    653
                Some(BreakpointCondition::RowCount { operator: _, value: _ }) => {
    654
                    // In real implementation, would check actual row count
    655
    0
                    Ok(true)
    656
                }
    657
                Some(BreakpointCondition::ExecutionTime { threshold_ms: _ }) => {
    658
                    // In real implementation, would check execution time
    659
    0
                    Ok(false)
    660
                }
    661
                Some(BreakpointCondition::MemoryUsage { threshold_mb: _ }) => {
    662
                    // In real implementation, would check memory usage
    663
    0
                    Ok(false)
    664
                }
    665
                Some(BreakpointCondition::DataValue { column: _, value: _ }) => {
    666
                    // In real implementation, would inspect actual data
    667
    0
                    Ok(false)
    668
                }
    669
    0
                Some(BreakpointCondition::OnError) => Ok(false),
    670
            }
    671
    0
        }
    672
    673
    0
        fn handle_breakpoint_hit(&self, stage_id: &str, breakpoint: &Breakpoint) -> Result<()> {
    674
            // Update breakpoint hit count
    675
    0
            let mut breakpoints = self.breakpoints
    676
    0
                .lock()
    677
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire breakpoints lock"))?;
    678
    679
    0
            if let Some(bp) = breakpoints.get_mut(stage_id) {
    680
    0
                bp.hit_count += 1;
    681
    0
            }
    682
    683
            // Update session state
    684
            {
    685
    0
                let mut state = self.session_state
    686
    0
                    .lock()
    687
    0
                    .map_err(|_| anyhow::anyhow!("Failed to acquire session state lock"))?;
    688
    0
                state.breakpoints_hit += 1;
    689
            }
    690
    691
            // Execute breakpoint actions
    692
    0
            for action in &breakpoint.actions {
    693
    0
                match action {
    694
    0
                    BreakpointAction::Pause => {
    695
    0
                        // In real implementation, would pause execution and wait for user input
    696
    0
                    }
    697
    0
                    BreakpointAction::Print(message) => {
    698
    0
                        println!("Breakpoint: {message}");
    699
    0
                    }
    700
                    BreakpointAction::Materialize => {
    701
    0
                        self.materialize_stage(stage_id)?;
    702
                    }
    703
    0
                    BreakpointAction::ComputeDiff => {
    704
    0
                        // Would compute diff with previous stage if available
    705
    0
                    }
    706
    0
                    BreakpointAction::Export { format: _, path: _ } => {
    707
    0
                        // Would export current data
    708
    0
                    }
    709
                }
    710
            }
    711
    712
    0
            self.record_event(
    713
    0
                EventType::BreakpointHit,
    714
    0
                stage_id.to_string(),
    715
    0
                HashMap::from([("hit_count".to_string(), breakpoint.hit_count.to_string())])
    716
    0
            )?;
    717
    718
    0
            Ok(())
    719
    0
        }
    720
    721
    0
        fn record_event(&self, event_type: EventType, stage_id: String, data: HashMap<String, String>) -> Result<()> {
    722
    0
            let event = ExecutionEvent {
    723
    0
                timestamp: std::time::SystemTime::now(),
    724
    0
                event_type,
    725
    0
                stage_id,
    726
    0
                data,
    727
    0
            };
    728
    729
    0
            let mut history = self.execution_history
    730
    0
                .lock()
    731
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire execution history lock"))?;
    732
    733
    0
            history.push_back(event);
    734
    735
            // Maintain history size limit
    736
    0
            while history.len() > self.config.max_history_events {
    737
    0
                history.pop_front();
    738
    0
            }
    739
    740
    0
            Ok(())
    741
    0
        }
    742
    743
    0
        fn compute_column_changes(&self, schema1: &DataSchema, schema2: &DataSchema) -> Vec<ColumnChange> {
    744
    0
            let mut changes = Vec::new();
    745
    746
            // Find added/removed columns (simplified implementation)
    747
    0
            let cols1: Vec<&str> = schema1.columns.iter().map(|c| c.name.as_str()).collect();
    748
    0
            let cols2: Vec<&str> = schema2.columns.iter().map(|c| c.name.as_str()).collect();
    749
    750
    0
            for col in &cols2 {
    751
    0
                if !cols1.contains(col) {
    752
    0
                    changes.push(ColumnChange::Added((*col).to_string()));
    753
    0
                }
    754
            }
    755
    756
    0
            for col in &cols1 {
    757
    0
                if !cols2.contains(col) {
    758
    0
                    changes.push(ColumnChange::Removed((*col).to_string()));
    759
    0
                }
    760
            }
    761
    762
    0
            changes
    763
    0
        }
    764
    765
    0
        fn compute_data_changes(&self, _data1: &[DataRow], _data2: &[DataRow]) -> Vec<DataChange> {
    766
            // Simplified implementation - in reality would compute detailed row-level diffs
    767
    0
            vec![DataChange::RowCountChanged]
    768
    0
        }
    769
    }
    770
    771
    /// Result of executing a pipeline stage
    772
    #[derive(Debug, Clone, PartialEq, Eq)]
    773
    pub enum StageExecutionResult {
    774
        /// Stage completed successfully
    775
        Completed,
    776
        /// Stage paused at breakpoint
    777
        Paused,
    778
        /// Stage failed with error
    779
        Failed(String),
    780
    }
    781
    782
    /// Diff between two pipeline stages
    783
    #[derive(Debug, Clone, Serialize, Deserialize)]
    784
    pub struct StageDiff {
    785
        /// First stage ID
    786
        pub stage1_id: String,
    787
    788
        /// Second stage ID
    789
        pub stage2_id: String,
    790
    791
        /// Difference in row count (stage2 - stage1)
    792
        pub row_count_diff: i64,
    793
    794
        /// Whether schema changed between stages
    795
        pub schema_changed: bool,
    796
    797
        /// Specific column changes
    798
        pub column_changes: Vec<ColumnChange>,
    799
    800
        /// Data-level changes
    801
        pub data_changes: Vec<DataChange>,
    802
    }
    803
    804
    /// Types of column changes between stages
    805
    #[derive(Debug, Clone, Serialize, Deserialize)]
    806
    pub enum ColumnChange {
    807
        /// Column was added
    808
        Added(String),
    809
        /// Column was removed
    810
        Removed(String),
    811
        /// Column type changed
    812
        TypeChanged(String, DataType, DataType),
    813
        /// Column renamed
    814
        Renamed(String, String),
    815
    }
    816
    817
    /// Types of data changes between stages
    818
    #[derive(Debug, Clone, Serialize, Deserialize)]
    819
    pub enum DataChange {
    820
        /// Row count changed
    821
        RowCountChanged,
    822
        /// Specific rows added
    823
        RowsAdded(Vec<usize>),
    824
        /// Specific rows removed
    825
        RowsRemoved(Vec<usize>),
    826
        /// Cell values modified
    827
        ValuesModified(Vec<(usize, usize)>), // (row, col) indices
    828
    }
    829
    830
    /// Complete debugging data export structure
    831
    #[derive(Debug, Clone, Serialize, Deserialize)]
    832
    pub struct DebugExport {
    833
        /// Current session status
    834
        pub session_status: SessionState,
    835
    836
        /// Execution history
    837
        pub execution_history: Vec<ExecutionEvent>,
    838
    839
        /// Performance metrics
    840
        pub stage_metrics: HashMap<String, StageMetrics>,
    841
    842
        /// Materialized data
    843
        pub materialized_data: HashMap<String, MaterializedFrame>,
    844
    }
    845
    846
    impl Default for SessionState {
    847
    0
        fn default() -> Self {
    848
    0
            Self {
    849
    0
                active: false,
    850
    0
                current_stage: None,
    851
    0
                session_start: std::time::SystemTime::now(),
    852
    0
                total_execution_time: Duration::from_secs(0),
    853
    0
                breakpoints_hit: 0,
    854
    0
                metadata: HashMap::new(),
    855
    0
            }
    856
    0
        }
    857
    }
    858
    859
    impl fmt::Display for StageType {
    860
    0
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    861
    0
            match self {
    862
    0
                Self::Load => write!(f, "Load"),
    863
    0
                Self::Filter => write!(f, "Filter"),
    864
    0
                Self::Select => write!(f, "Select"),
    865
    0
                Self::Map => write!(f, "Map"),
    866
    0
                Self::GroupBy => write!(f, "GroupBy"),
    867
    0
                Self::Aggregate => write!(f, "Aggregate"),
    868
    0
                Self::Join => write!(f, "Join"),
    869
    0
                Self::Sort => write!(f, "Sort"),
    870
    0
                Self::Window => write!(f, "Window"),
    871
    0
                Self::Union => write!(f, "Union"),
    872
    0
                Self::Custom(name) => write!(f, "Custom({name})"),
    873
            }
    874
    0
        }
    875
    }
    876
    877
    impl fmt::Display for StageStatus {
    878
    0
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    879
    0
            match self {
    880
    0
                Self::Pending => write!(f, "Pending"),
    881
    0
                Self::Running => write!(f, "Running"),
    882
    0
                Self::Completed => write!(f, "Completed"),
    883
    0
                Self::Failed(err) => write!(f, "Failed: {err}"),
    884
    0
                Self::Cancelled => write!(f, "Cancelled"),
    885
    0
                Self::Paused => write!(f, "Paused"),
    886
            }
    887
    0
        }
    888
    }
    889
    890
    impl fmt::Display for DataType {
    891
    0
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    892
    0
            match self {
    893
    0
                Self::Boolean => write!(f, "Boolean"),
    894
    0
                Self::Integer => write!(f, "Integer"),
    895
    0
                Self::Float => write!(f, "Float"),
    896
    0
                Self::String => write!(f, "String"),
    897
    0
                Self::DateTime => write!(f, "DateTime"),
    898
    0
                Self::Array(inner) => write!(f, "Array<{inner}>"),
    899
    0
                Self::Struct(fields) => {
    900
    0
                    write!(f, "Struct{{")?;
    901
    0
                    for (i, (name, dtype)) in fields.iter().enumerate() {
    902
    0
                        if i > 0 { write!(f, ", ")?; }
    903
    0
                        write!(f, "{name}: {dtype}")?;
    904
                    }
    905
    0
                    write!(f, "}}")
    906
                }
    907
            }
    908
    0
        }
    909
    }
    910
    911
    impl fmt::Display for DataValue {
    912
    0
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    913
    0
            match self {
    914
    0
                Self::Boolean(b) => write!(f, "{b}"),
    915
    0
                Self::Integer(i) => write!(f, "{i}"),
    916
    0
                Self::Float(fl) => write!(f, "{fl}"),
    917
    0
                Self::String(s) => write!(f, "\"{s}\""),
    918
    0
                Self::Null => write!(f, "null"),
    919
    0
                Self::Array(values) => {
    920
    0
                    write!(f, "[")?;
    921
    0
                    for (i, value) in values.iter().enumerate() {
    922
    0
                        if i > 0 { write!(f, ", ")?; }
    923
    0
                        write!(f, "{value}")?;
    924
                    }
    925
    0
                    write!(f, "]")
    926
                }
    927
            }
    928
    0
        }
    929
    }