4403513179

Build Type

push

github

Committed by GitHub

Commit Message

feat: Implement schema support for postgres (#1070)

Run Details

190 of 190 new or added lines in 28 files covered. (100.0%)

28368 of 38589 relevant lines covered (73.51%)

96135.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.23

/dozer-types/src/models/source.rs

use super::connection::Connection;
use serde::{ser::SerializeStruct, Deserialize, Serialize};

#[derive(Deserialize, Eq, PartialEq, Clone, ::prost::Message)]
pub struct Source {
    #[prost(string, tag = "1")]
    /// name of the source - to distinguish between multiple sources; Type: String
    pub name: String,
    #[prost(string, tag = "2")]
    /// name of the table in source database; Type: String
    pub table_name: String,
    #[prost(string, repeated, tag = "3")]
    /// list of columns gonna be used in the source table; Type: String[]
    pub columns: Vec<String>,
    #[prost(message, tag = "4")]
    #[serde(skip_deserializing)]
    /// reference to pre-defined connection name - syntax: `!Ref <connection_name>`; Type: `Ref!` tag
    pub connection: Option<Connection>,
    /// name of schema source database; Type: String
    #[prost(string, optional, tag = "5")]
    #[serde(default = "default_schema")]
    pub schema: Option<String>,
    #[prost(oneof = "RefreshConfig", tags = "7")]
    #[serde(default = "default_refresh_config")]
    #[serde(skip_serializing_if = "Option::is_none")]
    /// setting for how to refresh the data; Default: RealTime
    pub refresh_config: Option<RefreshConfig>,
}
fn default_schema() -> Option<String> {
    Some("public".to_string())
}
fn default_refresh_config() -> Option<RefreshConfig> {
    Some(RefreshConfig::default())
}
impl Serialize for Source {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let mut state = serializer.serialize_struct("Source", 5)?;
        state.serialize_field("name", &self.name)?;
        state.serialize_field("table_name", &self.table_name)?;
        state.serialize_field("columns", &self.columns)?;
        state.serialize_field("schema", &self.schema)?;
        state.serialize_field(
            "connection",
            &Value::Ref(self.connection.to_owned().unwrap_or_default().name),
        )?;
        state.serialize_field("refresh_config", &self.refresh_config)?;
        state.end()
    }
}
#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]
pub enum Value {
    Ref(String),
}

#[derive(Serialize, Deserialize, Eq, PartialEq, Clone, Debug)]
pub enum HistoryType {
    Master(MasterHistoryConfig),
    Transactional(TransactionalHistoryConfig),
}
#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]
pub enum MasterHistoryConfig {
    AppendOnly {
        unique_key_field: String,
        open_date_field: String,
        closed_date_field: String,
    },
    Overwrite,
}

#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]
pub enum TransactionalHistoryConfig {
    RetainPartial {
        timestamp_field: String,
        retention_period: u32,
    },
}
#[derive(Serialize, Deserialize, Eq, PartialEq, Clone, ::prost::Oneof)]
pub enum RefreshConfig {
    // Hour { minute: u32 },
    // Day { time: String },
    // CronExpression { expression: String },
    #[prost(message, tag = "7")]
    RealTime(RealTimeConfig),
}
impl Default for RefreshConfig {
    fn default() -> Self {
        RefreshConfig::RealTime(RealTimeConfig {})
    }
}
#[derive(Serialize, Deserialize, Eq, PartialEq, Clone, ::prost::Message)]
pub struct RealTimeConfig {}

1	use super::connection::Connection;
2	use serde::{ser::SerializeStruct, Deserialize, Serialize};
3
4	#[derive(Deserialize, Eq, PartialEq, Clone, ::prost::Message)]	1,388✔
5	pub struct Source {
6	#[prost(string, tag = "1")]
7	/// name of the source - to distinguish between multiple sources; Type: String
8	pub name: String,
9	#[prost(string, tag = "2")]
10	/// name of the table in source database; Type: String
11	pub table_name: String,
12	#[prost(string, repeated, tag = "3")]
13	/// list of columns gonna be used in the source table; Type: String[]
14	pub columns: Vec<String>,
15	#[prost(message, tag = "4")]
16	#[serde(skip_deserializing)]
17	/// reference to pre-defined connection name - syntax: `!Ref <connection_name>`; Type: `Ref!` tag
18	pub connection: Option<Connection>,
19	/// name of schema source database; Type: String
20	#[prost(string, optional, tag = "5")]
21	#[serde(default = "default_schema")]
22	pub schema: Option<String>,
23	#[prost(oneof = "RefreshConfig", tags = "7")]
24	#[serde(default = "default_refresh_config")]
25	#[serde(skip_serializing_if = "Option::is_none")]	×
26	/// setting for how to refresh the data; Default: RealTime	×
27	pub refresh_config: Option<RefreshConfig>,	×
28	}
29	fn default_schema() -> Option<String> {	421✔
30	Some("public".to_string())	421✔
31	}	421✔
32	fn default_refresh_config() -> Option<RefreshConfig> {	431✔
33	Some(RefreshConfig::default())	431✔
34	}	431✔
35	impl Serialize for Source {	×
36	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>	×
37	where	×
38	S: serde::Serializer,	×
39	{	×
40	let mut state = serializer.serialize_struct("Source", 5)?;	×
41	state.serialize_field("name", &self.name)?;	×
42	state.serialize_field("table_name", &self.table_name)?;	×
43	state.serialize_field("columns", &self.columns)?;	×
44	state.serialize_field("schema", &self.schema)?;	×
45	state.serialize_field(	×
46	"connection",	×
47	&Value::Ref(self.connection.to_owned().unwrap_or_default().name),	×
48	)?;	×
49	state.serialize_field("refresh_config", &self.refresh_config)?;	×
50	state.end()	×
51	}	×
52	}
53	#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]	346✔
54	pub enum Value {
55	Ref(String),	×
56	}
57
58	#[derive(Serialize, Deserialize, Eq, PartialEq, Clone, Debug)]	×
59	pub enum HistoryType {
60	Master(MasterHistoryConfig),
61	Transactional(TransactionalHistoryConfig),
62	}
63	#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]	×
64	pub enum MasterHistoryConfig {
65	AppendOnly {	×
66	unique_key_field: String,
67	open_date_field: String,
68	closed_date_field: String,
69	},
70	Overwrite,
71	}
72		×
73	#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]	×
74	pub enum TransactionalHistoryConfig {
75	RetainPartial {
76	timestamp_field: String,
77	retention_period: u32,
78	},
79	}
80	#[derive(Serialize, Deserialize, Eq, PartialEq, Clone, ::prost::Oneof)]	152✔
81	pub enum RefreshConfig {	×
82	// Hour { minute: u32 },	×
83	// Day { time: String },	×
84	// CronExpression { expression: String },
85	#[prost(message, tag = "7")]	×
86	RealTime(RealTimeConfig),
87	}
88	impl Default for RefreshConfig {
89	fn default() -> Self {	431✔
90	RefreshConfig::RealTime(RealTimeConfig {})	431✔
91	}	431✔
92	}
93	#[derive(Serialize, Deserialize, Eq, PartialEq, Clone, ::prost::Message)]	152✔
94	pub struct RealTimeConfig {}

getdozer / dozer / 4403513179

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous