pg2parquet 0.1.0

Command line tool for exporting PostgreSQL tables or queries into Parquet files
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

use std::{ops::Sub, net::IpAddr};

use bit_vec::BitVec;
use chrono::{Datelike, format::Fixed};
use parquet::data_type::{ByteArray, FixedLenByteArray, Int64Type};
use pg_bigdecimal::{PgNumeric, BigDecimal};

use crate::pg_custom_types::PgEnum;

pub trait MyFrom<T> {
	fn my_from(t: T) -> Self;
}

impl<T> MyFrom<T> for T {
	fn my_from(t: T) -> Self {
		t
	}
}

impl MyFrom<i32> for i64 {
	fn my_from(t: i32) -> Self {
		t as i64
	}
}

impl MyFrom<i16> for i32 {
	fn my_from(t: i16) -> Self {
		t as i32
	}
}

impl MyFrom<i8> for i32 {
	fn my_from(t: i8) -> Self {
		t as i32
	}
}

impl MyFrom<u32> for i32 {
	fn my_from(t: u32) -> Self {
		t as i32
	}
}

impl MyFrom<Vec<u8>> for ByteArray {
	fn my_from(t: Vec<u8>) -> Self {
		ByteArray::from(t)
	}
}
impl MyFrom<String> for ByteArray {
	fn my_from(t: String) -> Self {
		ByteArray::from(t.into_bytes())
	}
}

impl MyFrom<&str> for ByteArray {
	fn my_from(t: &str) -> Self {
		ByteArray::from(t)
	}
}

impl MyFrom<chrono::DateTime<chrono::Utc>> for i64 {
	fn my_from(t: chrono::DateTime<chrono::Utc>) -> Self {
		t.timestamp_micros()
	}
}

impl MyFrom<chrono::NaiveDateTime> for i64 {
	fn my_from(t: chrono::NaiveDateTime) -> Self {
		t.timestamp_micros()
	}
}

impl MyFrom<chrono::NaiveDate> for i32 {
	fn my_from(t: chrono::NaiveDate) -> Self {
		// number of days since 1970-01-01
		t.sub(chrono::NaiveDate::from_ymd_opt(1970, 1, 1).unwrap()).num_days() as i32
	}
}

impl MyFrom<chrono::NaiveTime> for i64 {
	fn my_from(t: chrono::NaiveTime) -> Self {
		// number of seconds since 00:00:00
		t.signed_duration_since(chrono::NaiveTime::from_hms_opt(0, 0, 0).unwrap()).num_microseconds().unwrap()
	}
}

impl MyFrom<uuid::Uuid> for FixedLenByteArray {
	fn my_from(t: uuid::Uuid) -> Self {
		FixedLenByteArray::from(t.as_bytes().to_vec())
	}
}

impl<'a> MyFrom<PgEnum> for ByteArray {
	fn my_from(t: PgEnum) -> Self {
		ByteArray::from(t.name.into_bytes())
	}
}

impl<'a> MyFrom<eui48::MacAddress> for ByteArray {
	fn my_from(t: eui48::MacAddress) -> Self {
		let str = t.to_hex_string();
		ByteArray::from(str.into_bytes())
	}
}
impl<'a> MyFrom<eui48::MacAddress> for FixedLenByteArray {
	fn my_from(t: eui48::MacAddress) -> Self {
		let b = t.as_bytes();
		FixedLenByteArray::from(b.to_vec())
	}
}
impl<'a> MyFrom<eui48::MacAddress> for i64 {
	fn my_from(t: eui48::MacAddress) -> Self {
		let mut b = [0u8; 8];
		b[0..6].copy_from_slice(t.as_bytes());
		i64::from_be_bytes(b)
	}
}
impl<'a> MyFrom<IpAddr> for ByteArray {
	fn my_from(t: IpAddr) -> Self {
		let str = t.to_string();
		ByteArray::from(str.into_bytes())
	}
}
impl<'a> MyFrom<BitVec> for ByteArray {
	fn my_from(t: BitVec) -> Self {
		// this format should be easiest to work with, and parquet should compress it anyway
		let str = t.iter().map(|b| if b { '1' } else { '0' }).collect::<String>();
		ByteArray::from(str.into_bytes())
	}
}