elefant-tools 0.0.8

A library for doing things like pg_dump and pg_restore, with extra features, and probably more bugs.
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

use crate::postgres_client_wrapper::QueryResult;
use elefant_client::Interval;

pub struct ContinuousAggregateResult {
    // pub hypertable_schema: String,
    // pub hypertable_name: String,
    pub view_schema: String,
    pub view_name: String,
    // pub materialized_only: bool,
    pub view_definition: String,
    pub refresh_interval: Option<Interval>,
    pub refresh_start_offset: Option<Interval>,
    pub refresh_end_offset: Option<Interval>,
    pub compression_enabled: bool,
    pub compress_job_interval: Option<Interval>,
    pub compress_after: Option<Interval>,
    pub compress_order_by: Option<Vec<String>>,
    pub compress_order_by_desc: Option<Vec<bool>>,
    pub compress_order_by_nulls_first: Option<Vec<bool>>,
    pub compress_segment_by: Option<Vec<String>>,
    pub compress_chunk_time_interval: Option<Interval>,
    pub retention_schedule_interval: Option<Interval>,
    pub retention_drop_after: Option<Interval>,
}

impl<'a> elefant_client::FromSqlRow<'a> for ContinuousAggregateResult {
    fn from_sql_row(
        row: &'a elefant_client::PostgresDataRow<'_, '_>,
    ) -> Result<Self, elefant_client::ElefantClientError> {
        Ok(ContinuousAggregateResult {
            // hypertable_schema: row.get(0)?,
            // hypertable_name: row.get(1)?,
            view_schema: row.get(2)?,
            view_name: row.get(3)?,
            // materialized_only: row.get(4)?,
            view_definition: row.get(5)?,
            refresh_interval: row.get(6)?,
            refresh_start_offset: row.get(7)?,
            refresh_end_offset: row.get(8)?,
            compression_enabled: row.get(9)?,
            compress_job_interval: row.get(10)?,
            compress_after: row.get(11)?,
            compress_order_by: row.get(12)?,
            compress_order_by_desc: row.get(13)?,
            compress_order_by_nulls_first: row.get(14)?,
            compress_segment_by: row.get(15)?,
            compress_chunk_time_interval: row.get(16)?,
            retention_schedule_interval: row.get(17)?,
            retention_drop_after: row.get(18)?,
        })
    }
}

//language=postgresql
pub(in crate::schema_reader) const QUERY: &str = r#"

SELECT ht.schema_name                                       AS hypertable_schema,
       ht.table_name                                        AS hypertable_name,
       cagg.user_view_schema                                AS view_schema,
       cagg.user_view_name                                  AS view_name,
       cagg.materialized_only,
       pg_get_viewdef(c.oid)                                AS view_definition,
       refresh_job.schedule_interval                        as refresh_interval,
       (refresh_job.config ->> 'start_offset')::interval    as refresh_start_offset,
       (refresh_job.config ->> 'end_offset')::interval      as refresh_end_offset,
       CASE
           WHEN mat_ht.compressed_hypertable_id IS NOT NULL THEN true
           ELSE false
           END                                              AS compression_enabled,
       compress_job.schedule_interval                       as compress_job_interval,
       (compress_job.config ->> 'compress_after')::interval as compress_after,
       cs.orderby                                           as compress_orderby,
       cs.orderby_desc                                      as compress_orderby_desc,
       cs.orderby_nullsfirst                                as compress_orderby_nullsfirst,
       cs.segmentby                                         as compress_segmentby,
        _timescaledb_functions.to_interval(dim.compress_interval_length) as compress_chunk_time_interval,
        retention_job.schedule_interval as retention_schedule_interval,
        (retention_job.config->>'drop_after')::interval as retention_drop_after
FROM _timescaledb_catalog.continuous_agg cagg
         join _timescaledb_catalog.hypertable ht on cagg.raw_hypertable_id = ht.id
         join _timescaledb_catalog.hypertable mat_ht on cagg.mat_hypertable_id = mat_ht.id
         join pg_class c on c.relname = cagg.direct_view_name and c.relkind = 'v'
         join pg_namespace n on n.oid = c.relnamespace and n.nspname = cagg.direct_view_schema
        left join _timescaledb_catalog.dimension dim on mat_ht.id = dim.hypertable_id and dim.compress_interval_length is not null
         left join _timescaledb_config.bgw_job refresh_job on refresh_job.hypertable_id = mat_ht.id and
                                                              refresh_job.proc_name =
                                                              'policy_refresh_continuous_aggregate' and
                                                              refresh_job.proc_schema = '_timescaledb_functions'
         left join _timescaledb_config.bgw_job compress_job
                   on compress_job.hypertable_id = mat_ht.id and compress_job.proc_name = 'policy_compression' and
                      compress_job.proc_schema = '_timescaledb_functions'
         left join _timescaledb_catalog.compression_settings cs
                   on cs.relid = (mat_ht.schema_name || '.' || mat_ht.table_name)::regclass
left join _timescaledb_config.bgw_job retention_job on retention_job.hypertable_id = mat_ht.id and retention_job.proc_name = 'policy_retention' and retention_job.proc_schema = '_timescaledb_functions';
"#;

impl QueryResult for ContinuousAggregateResult {
    fn query(_version: i32) -> &'static str {
        QUERY
    }
}