pgdrift 0.1.2

Command-line tool for detecting schema drift in PostgreSQL JSONB columns
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use crate::output::{ColumnScanResult, OutputFormat, ScanAllResult};
use anyhow::{Context, Result};
use pgdrift_core::analyzer::JsonAnalyzer;
use pgdrift_core::drift::{DriftConfig, DriftIssue, Severity, detect_drift};
use pgdrift_db::{ConnectionPool, Sampler, discover_jsonb_columns};

/// Run scan-all command to analyze all JSONB columns in the given DB
pub async fn run(database_url: &str, sample_size: usize, format: OutputFormat) -> Result<()> {
    let conn = ConnectionPool::new(database_url)
        .await
        .context("Failed to connect to the database")?;

    conn.test_connection()
        .await
        .context("Failed to test the database connection")?;

    let columns = discover_jsonb_columns(conn.pool())
        .await
        .context("Failed to discover JSONB columns")?;

    if columns.is_empty() {
        println!("No JSONB columns found in the database.");
        return Ok(());
    }

    println!(
        "Discovered {} JSONB columns. Starting analysis...\n",
        columns.len()
    );

    let mut column_results = Vec::new();
    let config = DriftConfig::default();

    for col in &columns {
        println!(
            "Analyzing column: {}.{} (table: {})",
            col.schema, col.column, col.table
        );

        match analyze_column(
            conn.pool(),
            &col.schema,
            &col.table,
            &col.column,
            sample_size,
            &config,
        )
        .await
        {
            Ok((samples_analyzed, drift_issues)) => {
                let critical = drift_issues
                    .iter()
                    .filter(|i| i.severity() == Severity::Critical)
                    .count();
                let warning = drift_issues
                    .iter()
                    .filter(|i| i.severity() == Severity::Warning)
                    .count();
                let info = drift_issues
                    .iter()
                    .filter(|i| i.severity() == Severity::Info)
                    .count();

                println!(
                    "Analysis complete for {}.{}.{} - Samples Analyzed: {}, Issues Found: {} (Critical: {}, Warning: {}, Info: {})\n",
                    col.schema,
                    col.table,
                    col.column,
                    samples_analyzed,
                    drift_issues.len(),
                    critical,
                    warning,
                    info
                );

                column_results.push(ColumnScanResult {
                    schema: col.schema.clone(),
                    table: col.table.clone(),
                    column: col.column.clone(),
                    samples_analyzed: samples_analyzed as u64,
                    drift_issues,
                });
            }
            Err(e) => {
                eprintln!(
                    "Error analyzing column {}.{}.{}: {}\n",
                    col.schema, col.table, col.column, e
                );
                // Continue with next column even if there's an error
                column_results.push(ColumnScanResult {
                    schema: col.schema.clone(),
                    table: col.table.clone(),
                    column: col.column.clone(),
                    samples_analyzed: 0,
                    drift_issues: vec![],
                });
            }
        }
    }

    let result = ScanAllResult {
        total_columns: columns.len(),
        column_results,
    };

    crate::output::print_scan_all_summary(&result, &format)?;

    Ok(())
}

async fn analyze_column(
    pool: &sqlx::PgPool,
    schema: &str,
    table: &str,
    column: &str,
    sample_size: usize,
    config: &DriftConfig,
) -> Result<(usize, Vec<DriftIssue>)> {
    let sampler = Sampler::new(pool, schema, table, None, sample_size)
        .await
        .context("Failed to create sampler")?
        .show_progress(false);

    let samples = sampler
        .sample(pool, schema, table, column)
        .await
        .context("Failed to sample data")?;

    if samples.is_empty() {
        anyhow::bail!("No samples found in the column");
    }

    let mut analyzer = JsonAnalyzer::new();
    for sample in &samples {
        analyzer.analyze(sample);
    }
    let stats = analyzer.finalize();
    let drift_issues = detect_drift(&stats, config);

    Ok((samples.len(), drift_issues))
}