hinge 0.1.0

SQL-native ELT engine — dependency graph resolved automatically from FROM/JOIN clauses, parallel execution, single binary
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

use sqlx::PgConnection;
use crate::{Asset, AssetKind, ExecutorError};

pub async fn run_in_transaction(
    conn:  &mut PgConnection,
    asset: &Asset,
) -> Result<u64, ExecutorError> {
    let schema = asset.reference().schema();
    let name   = asset.reference().name();
    let source = asset.source().as_str();

    // advisory lock
    sqlx::query("SELECT pg_advisory_xact_lock($1)")
        .bind(lock_key(asset))
        .execute(&mut *conn)
        .await
        .map_err(|e| ExecutorError::Run {
            asset:   asset.to_string(),
            message: e.to_string(),
        })?;
    tracing::debug!(asset = %asset, "lock acquired");

    // save existing indexes
    let saved_indexes: Vec<String> = match asset.kind() {
        AssetKind::View => vec![],
        _               => save_indexes(&mut *conn, asset).await?,
    };

    // drop whatever exists
    drop_any(&mut *conn, asset).await?;

    // create
    let rows_affected = match asset.kind() {
        AssetKind::View => {
            let sql = format!(r#"CREATE VIEW "{schema}"."{name}" AS {source}"#);
            exec(&mut *conn, asset, &sql).await?
        }
        AssetKind::Table => {
            let sql = format!(r#"CREATE TABLE "{schema}"."{name}" AS {source}"#);
            exec(&mut *conn, asset, &sql).await?
        }
        AssetKind::MaterializedView => {
            let sql = format!(
                r#"CREATE MATERIALIZED VIEW "{schema}"."{name}" AS {source} WITH DATA"#
            );
            exec(&mut *conn, asset, &sql).await?
        }
    };

    // restore indexes
    let n = saved_indexes.len();
    for def in &saved_indexes {
        exec(&mut *conn, asset, def).await?;
    }
    if n > 0 {
        tracing::debug!(asset = %asset, n_indexes = n, "indexes restored");
    }

    Ok(rows_affected)
}


async fn exec(conn: &mut PgConnection, asset: &Asset, sql: &str) -> Result<u64, ExecutorError> {
    sqlx::query(sql)
        .execute(conn)
        .await
        .map(|r| r.rows_affected())
        .map_err(|e| ExecutorError::Run {
            asset:   asset.to_string(),
            message: e.to_string(),
        })
}

async fn drop_any(conn: &mut PgConnection, asset: &Asset) -> Result<(), ExecutorError> {
    let schema = asset.reference().schema();
    let name   = asset.reference().name();

    let kind: Option<String> = sqlx::query_scalar(
        r#"
        SELECT c.relkind::text
        FROM   pg_class     c
        JOIN   pg_namespace n ON n.oid = c.relnamespace
        WHERE  n.nspname = $1 AND c.relname = $2
        "#,
    )
        .bind(schema)
        .bind(name)
        .fetch_optional(&mut *conn)
        .await
        .map_err(|e| ExecutorError::Run {
            asset:   asset.to_string(),
            message: e.to_string(),
        })?;

    let drop_sql = match kind.as_deref() {
        Some("r") => format!(r#"DROP TABLE             "{schema}"."{name}""#),
        Some("v") => format!(r#"DROP VIEW              "{schema}"."{name}""#),
        Some("m") => format!(r#"DROP MATERIALIZED VIEW "{schema}"."{name}""#),
        _         => return Ok(()),
    };

    exec(conn, asset, &drop_sql).await?;
    Ok(())
}

async fn save_indexes(
    conn:  &mut PgConnection,
    asset: &Asset,
) -> Result<Vec<String>, ExecutorError> {
    let schema = asset.reference().schema();
    let name   = asset.reference().name();

    sqlx::query_scalar(
        r#"SELECT indexdef FROM pg_indexes WHERE schemaname = $1 AND tablename = $2"#,
    )
        .bind(schema)
        .bind(name)
        .fetch_all(&mut *conn)
        .await
        .map_err(|e| ExecutorError::Run {
            asset:   asset.to_string(),
            message: e.to_string(),
        })
}

fn lock_key(asset: &Asset) -> i64 {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    let mut h = DefaultHasher::new();
    asset.reference().schema().hash(&mut h);
    asset.reference().name().hash(&mut h);
    h.finish() as i64
}