4195006582

Build # 4195006582

Build Type

github

Committed by GitHub

Commit Message

Merge 14ceac9a6 into dc166625f

Pull Request Pull Request #918: ObjectStore - Reduce code duplication & more clone prevention

Run Details

177 of 177 new or added lines in 5 files covered. (100.0%)

24368 of 37519 relevant lines covered (64.95%)

43913.78 hits per line

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

0.0

/dozer-ingestion/src/connectors/object_store/schema_helper.rs

use crate::errors::ObjectStoreSchemaError;
use crate::errors::ObjectStoreSchemaError::DateConversionError;
use crate::errors::ObjectStoreSchemaError::DateTimeConversionError;
use crate::errors::ObjectStoreSchemaError::DurationConversionError;
use crate::errors::ObjectStoreSchemaError::FieldTypeNotSupported;
use crate::errors::ObjectStoreSchemaError::TimeConversionError;
use datafusion::arrow::array;
use datafusion::arrow::array::{Array, ArrayRef};
use datafusion::arrow::datatypes::{DataType, Field, TimeUnit};

use dozer_types::types::{Field as DozerField, FieldDefinition, FieldType, SourceDefinition};

macro_rules! make_from {
    ($array_type:ty, $column: ident, $row: ident) => {{
        let array = $column.as_any().downcast_ref::<$array_type>();

        if let Some(r) = array {
            let s: DozerField = if r.is_null($row.clone()) {
                DozerField::Null
            } else {
                DozerField::from(r.value($row.clone()))
            };

            Ok(s)
        } else {
            Ok(DozerField::Null)
        }
    }};
}

macro_rules! make_binary {
    ($array_type:ty, $column: ident, $row: ident) => {{
        let array = $column.as_any().downcast_ref::<$array_type>();

        if let Some(r) = array {
            let s: DozerField = if r.is_null($row.clone()) {
                DozerField::Null
            } else {
                DozerField::Binary(r.value($row.clone()).to_vec())
            };

            Ok(s)
        } else {
            Ok(DozerField::Null)
        }
    }};
}

macro_rules! make_timestamp {
    ($array_type:ty, $column: ident, $row: ident) => {{
        let array = $column.as_any().downcast_ref::<$array_type>();

        if let Some(r) = array {
            if r.is_null($row.clone()) {
                Ok(DozerField::Null)
            } else {
                r.value_as_datetime($row.clone())
                    .map(DozerField::from)
                    .map_or_else(|| Err(DateTimeConversionError), |v| Ok(DozerField::from(v)))
            }
        } else {
            Ok(DozerField::Null)
        }
    }};
}

macro_rules! make_date {
    ($array_type:ty, $column: ident, $row: ident) => {{
        let array = $column.as_any().downcast_ref::<$array_type>();

        if let Some(r) = array {
            if r.is_null($row.clone()) {
                Ok(DozerField::Null)
            } else {
                r.value_as_date($row.clone())
                    .map_or_else(|| Err(DateConversionError), |v| Ok(DozerField::from(v)))
            }
        } else {
            Ok(DozerField::Null)
        }
    }};
}

macro_rules! make_time {
    ($array_type:ty, $column: ident, $row: ident) => {{
        let array = $column.as_any().downcast_ref::<$array_type>();

        if let Some(r) = array {
            if r.is_null($row.clone()) {
                Ok(DozerField::Null)
            } else {
                r.value_as_time($row.clone())
                    .map_or_else(|| Err(TimeConversionError), |v| Ok(DozerField::from(v)))
            }
        } else {
            Ok(DozerField::Null)
        }
    }};
}

macro_rules! make_duration {
    ($array_type:ty, $column: ident, $row: ident) => {{
        let array = $column.as_any().downcast_ref::<$array_type>();

        if let Some(r) = array {
            if r.is_null($row.clone()) {
                Ok(DozerField::Null)
            } else {
                r.value_as_duration($row.clone()).map_or_else(
                    || Err(DurationConversionError),
                    |v| Ok(DozerField::from(v.num_nanoseconds().unwrap())),
                )
            }
        } else {
            Ok(DozerField::Null)
        }
    }};
}

pub fn map_schema_to_dozer<'a, I: Iterator<Item = &'a Field>>(
    fields_list: I,
) -> Result<Vec<FieldDefinition>, ObjectStoreSchemaError> {
    fields_list
        .map(|field| {
            let mapped_field_type = match field.data_type() {
                DataType::Boolean => FieldType::Boolean,
                DataType::Time32(_)
                | DataType::Time64(_)
                | DataType::Duration(_)
                | DataType::Interval(_)
                | DataType::Int8
                | DataType::Int16
                | DataType::Int32
                | DataType::Int64 => FieldType::Int,
                DataType::UInt8 | DataType::UInt16 | DataType::UInt32 | DataType::UInt64 => {
                    FieldType::UInt
                }
                DataType::Float16 | DataType::Float32 | DataType::Float64 => FieldType::Float,
                DataType::Timestamp(_, _) => FieldType::Timestamp,
                DataType::Date32 | DataType::Date64 => FieldType::Date,
                DataType::Binary | DataType::FixedSizeBinary(_) | DataType::LargeBinary => {
                    FieldType::Binary
                }
                DataType::Utf8 => FieldType::String,
                DataType::LargeUtf8 => FieldType::Text,
                // DataType::List(_) => {}
                // DataType::FixedSizeList(_, _) => {}
                // DataType::LargeList(_) => {}
                // DataType::Struct(_) => {}
                // DataType::Union(_, _, _) => {}
                // DataType::Dictionary(_, _) => {}
                // DataType::Decimal128(_, _) => {}
                // DataType::Decimal256(_, _) => {}
                // DataType::Map(_, _) => {}
                _ => return Err(FieldTypeNotSupported(field.name().clone())),
            };

            Ok(FieldDefinition {
                name: field.name().clone(),
                typ: mapped_field_type,
                nullable: field.is_nullable(),
                source: SourceDefinition::Dynamic,
            })
        })
        .collect()
}

pub fn map_value_to_dozer_field(
    column: &ArrayRef,
    row: &usize,
    column_name: &str,
) -> Result<DozerField, ObjectStoreSchemaError> {
    match column.data_type() {
        DataType::Null => Ok(DozerField::Null),
        DataType::Boolean => make_from!(array::BooleanArray, column, row),
        DataType::Int8 => make_from!(array::Int8Array, column, row),
        DataType::Int16 => make_from!(array::Int16Array, column, row),
        DataType::Int32 => make_from!(array::Int32Array, column, row),
        DataType::Int64 => make_from!(array::Int64Array, column, row),
        DataType::UInt8 => make_from!(array::UInt8Array, column, row),
        DataType::UInt16 => make_from!(array::UInt16Array, column, row),
        DataType::UInt32 => make_from!(array::UInt32Array, column, row),
        DataType::UInt64 => make_from!(array::UInt64Array, column, row),
        DataType::Float16 => make_from!(array::Float32Array, column, row),
        DataType::Float32 => make_from!(array::Float32Array, column, row),
        DataType::Float64 => make_from!(array::Float64Array, column, row),
        DataType::Timestamp(TimeUnit::Microsecond, _) => {
            make_timestamp!(array::TimestampMicrosecondArray, column, row)
        }
        DataType::Timestamp(TimeUnit::Millisecond, _) => {
            make_timestamp!(array::TimestampMillisecondArray, column, row)
        }
        DataType::Timestamp(TimeUnit::Nanosecond, _) => {
            make_timestamp!(array::TimestampNanosecondArray, column, row)
        }
        DataType::Timestamp(TimeUnit::Second, _) => {
            make_timestamp!(array::TimestampSecondArray, column, row)
        }
        DataType::Date32 => make_date!(array::Date32Array, column, row),
        DataType::Date64 => make_date!(array::Date64Array, column, row),
        DataType::Time32(TimeUnit::Millisecond) => {
            make_time!(array::Time32MillisecondArray, column, row)
        }
        DataType::Time32(TimeUnit::Second) => make_time!(array::Time32SecondArray, column, row),
        DataType::Time64(TimeUnit::Microsecond) => {
            make_time!(array::Time64MicrosecondArray, column, row)
        }
        DataType::Time64(TimeUnit::Nanosecond) => {
            make_time!(array::Time64NanosecondArray, column, row)
        }
        DataType::Duration(TimeUnit::Microsecond) => {
            make_duration!(array::DurationMicrosecondArray, column, row)
        }
        DataType::Duration(TimeUnit::Millisecond) => {
            make_duration!(array::DurationMillisecondArray, column, row)
        }
        DataType::Duration(TimeUnit::Nanosecond) => {
            make_duration!(array::DurationNanosecondArray, column, row)
        }
        DataType::Duration(TimeUnit::Second) => {
            make_duration!(array::DurationSecondArray, column, row)
        }
        DataType::Binary => make_binary!(array::BinaryArray, column, row),
        DataType::FixedSizeBinary(_) => make_binary!(array::FixedSizeBinaryArray, column, row),
        DataType::LargeBinary => make_binary!(array::LargeBinaryArray, column, row),
        DataType::Utf8 => make_from!(array::StringArray, column, row),
        DataType::LargeUtf8 => make_from!(array::StringArray, column, row),
        // DataType::Interval(TimeUnit::) => make_from!(array::BooleanArray, x, x0),
        // DataType::List(_) => {}
        // DataType::FixedSizeList(_, _) => {}
        // DataType::LargeList(_) => {}
        // DataType::Struct(_) => {}
        // DataType::Union(_, _, _) => {}
        // DataType::Dictionary(_, _) => {}
        // DataType::Decimal128(_, _) => {}
        // DataType::Decimal256(_, _) => {}
        // DataType::Map(_, _) => {}
        _ => Err(FieldTypeNotSupported(column_name.to_string())),
    }
}

1	use crate::errors::ObjectStoreSchemaError;
2	use crate::errors::ObjectStoreSchemaError::DateConversionError;
3	use crate::errors::ObjectStoreSchemaError::DateTimeConversionError;
4	use crate::errors::ObjectStoreSchemaError::DurationConversionError;
5	use crate::errors::ObjectStoreSchemaError::FieldTypeNotSupported;
6	use crate::errors::ObjectStoreSchemaError::TimeConversionError;
7	use datafusion::arrow::array;
8	use datafusion::arrow::array::{Array, ArrayRef};
9	use datafusion::arrow::datatypes::{DataType, Field, TimeUnit};
10
11	use dozer_types::types::{Field as DozerField, FieldDefinition, FieldType, SourceDefinition};
12
13	macro_rules! make_from {
14	($array_type:ty, $column: ident, $row: ident) => {{
15	let array = $column.as_any().downcast_ref::<$array_type>();
16
17	if let Some(r) = array {
18	let s: DozerField = if r.is_null($row.clone()) {
19	DozerField::Null
20	} else {
21	DozerField::from(r.value($row.clone()))
22	};
23
24	Ok(s)
25	} else {
26	Ok(DozerField::Null)
27	}
28	}};
29	}
30
31	macro_rules! make_binary {
32	($array_type:ty, $column: ident, $row: ident) => {{
33	let array = $column.as_any().downcast_ref::<$array_type>();
34
35	if let Some(r) = array {
36	let s: DozerField = if r.is_null($row.clone()) {
37	DozerField::Null
38	} else {
39	DozerField::Binary(r.value($row.clone()).to_vec())
40	};
41
42	Ok(s)
43	} else {
44	Ok(DozerField::Null)
45	}
46	}};
47	}
48
49	macro_rules! make_timestamp {
50	($array_type:ty, $column: ident, $row: ident) => {{
51	let array = $column.as_any().downcast_ref::<$array_type>();
52
53	if let Some(r) = array {
54	if r.is_null($row.clone()) {
55	Ok(DozerField::Null)
56	} else {
57	r.value_as_datetime($row.clone())
58	.map(DozerField::from)
59	.map_or_else(\|\| Err(DateTimeConversionError), \|v\| Ok(DozerField::from(v)))
60	}
61	} else {
62	Ok(DozerField::Null)
63	}
64	}};
65	}
66
67	macro_rules! make_date {
68	($array_type:ty, $column: ident, $row: ident) => {{
69	let array = $column.as_any().downcast_ref::<$array_type>();
70
71	if let Some(r) = array {
72	if r.is_null($row.clone()) {
73	Ok(DozerField::Null)
74	} else {
75	r.value_as_date($row.clone())
76	.map_or_else(\|\| Err(DateConversionError), \|v\| Ok(DozerField::from(v)))
77	}
78	} else {
79	Ok(DozerField::Null)
80	}
81	}};
82	}
83
84	macro_rules! make_time {
85	($array_type:ty, $column: ident, $row: ident) => {{
86	let array = $column.as_any().downcast_ref::<$array_type>();
87
88	if let Some(r) = array {
89	if r.is_null($row.clone()) {
90	Ok(DozerField::Null)
91	} else {
92	r.value_as_time($row.clone())
93	.map_or_else(\|\| Err(TimeConversionError), \|v\| Ok(DozerField::from(v)))
94	}
95	} else {
96	Ok(DozerField::Null)
97	}
98	}};
99	}
100
101	macro_rules! make_duration {
102	($array_type:ty, $column: ident, $row: ident) => {{
103	let array = $column.as_any().downcast_ref::<$array_type>();
104
105	if let Some(r) = array {
106	if r.is_null($row.clone()) {
107	Ok(DozerField::Null)
108	} else {
109	r.value_as_duration($row.clone()).map_or_else(
110	\|\| Err(DurationConversionError),
111	\|v\| Ok(DozerField::from(v.num_nanoseconds().unwrap())),
112	)
113	}
114	} else {
115	Ok(DozerField::Null)
116	}
117	}};
118	}
119
120	pub fn map_schema_to_dozer<'a, I: Iterator<Item = &'a Field>>(	×
121	fields_list: I,	×
122	) -> Result<Vec<FieldDefinition>, ObjectStoreSchemaError> {	×
123	fields_list	×
124	.map(\|field\| {	×
125	let mapped_field_type = match field.data_type() {	×
126	DataType::Boolean => FieldType::Boolean,	×
127	DataType::Time32(_)	×
128	\| DataType::Time64(_)	×
129	\| DataType::Duration(_)	×
130	\| DataType::Interval(_)	×
131	\| DataType::Int8	×
132	\| DataType::Int16	×
133	\| DataType::Int32	×
134	\| DataType::Int64 => FieldType::Int,	×
135	DataType::UInt8 \| DataType::UInt16 \| DataType::UInt32 \| DataType::UInt64 => {	×
136	FieldType::UInt	×
137	}	×
138	DataType::Float16 \| DataType::Float32 \| DataType::Float64 => FieldType::Float,	×
139	DataType::Timestamp(_, _) => FieldType::Timestamp,	×
140	DataType::Date32 \| DataType::Date64 => FieldType::Date,	×
141	DataType::Binary \| DataType::FixedSizeBinary(_) \| DataType::LargeBinary => {	×
142	FieldType::Binary	×
143	}	×
144	DataType::Utf8 => FieldType::String,	×
145	DataType::LargeUtf8 => FieldType::Text,	×
146	// DataType::List(_) => {}	×
147	// DataType::FixedSizeList(_, _) => {}	×
148	// DataType::LargeList(_) => {}	×
149	// DataType::Struct(_) => {}	×
150	// DataType::Union(_, _, _) => {}
151	// DataType::Dictionary(_, _) => {}
152	// DataType::Decimal128(_, _) => {}
153	// DataType::Decimal256(_, _) => {}
154	// DataType::Map(_, _) => {}
155	_ => return Err(FieldTypeNotSupported(field.name().clone())),	×
156	};
157
158	Ok(FieldDefinition {	×
159	name: field.name().clone(),	×
160	typ: mapped_field_type,	×
161	nullable: field.is_nullable(),	×
162	source: SourceDefinition::Dynamic,	×
163	})	×
164	})	×
165	.collect()	×
166	}	×
167		×
168	pub fn map_value_to_dozer_field(
169	column: &ArrayRef,
170	row: &usize,	×
171	column_name: &str,	×
172	) -> Result<DozerField, ObjectStoreSchemaError> {
173	match column.data_type() {	×
174	DataType::Null => Ok(DozerField::Null),	×
175	DataType::Boolean => make_from!(array::BooleanArray, column, row),	×
176	DataType::Int8 => make_from!(array::Int8Array, column, row),	×
177	DataType::Int16 => make_from!(array::Int16Array, column, row),	×
178	DataType::Int32 => make_from!(array::Int32Array, column, row),	×
179	DataType::Int64 => make_from!(array::Int64Array, column, row),	×
180	DataType::UInt8 => make_from!(array::UInt8Array, column, row),	×
181	DataType::UInt16 => make_from!(array::UInt16Array, column, row),	×
182	DataType::UInt32 => make_from!(array::UInt32Array, column, row),	×
183	DataType::UInt64 => make_from!(array::UInt64Array, column, row),	×
184	DataType::Float16 => make_from!(array::Float32Array, column, row),	×
185	DataType::Float32 => make_from!(array::Float32Array, column, row),	×
186	DataType::Float64 => make_from!(array::Float64Array, column, row),	×
187	DataType::Timestamp(TimeUnit::Microsecond, _) => {	×
188	make_timestamp!(array::TimestampMicrosecondArray, column, row)	×
189	}	×
190	DataType::Timestamp(TimeUnit::Millisecond, _) => {	×
191	make_timestamp!(array::TimestampMillisecondArray, column, row)	×
192	}
193	DataType::Timestamp(TimeUnit::Nanosecond, _) => {	×
194	make_timestamp!(array::TimestampNanosecondArray, column, row)	×
195	}
196	DataType::Timestamp(TimeUnit::Second, _) => {	×
197	make_timestamp!(array::TimestampSecondArray, column, row)	×
198	}
199	DataType::Date32 => make_date!(array::Date32Array, column, row),	×
200	DataType::Date64 => make_date!(array::Date64Array, column, row),	×
201	DataType::Time32(TimeUnit::Millisecond) => {
202	make_time!(array::Time32MillisecondArray, column, row)	×
203	}
204	DataType::Time32(TimeUnit::Second) => make_time!(array::Time32SecondArray, column, row),	×
205	DataType::Time64(TimeUnit::Microsecond) => {	×
206	make_time!(array::Time64MicrosecondArray, column, row)	×
207	}	×
208	DataType::Time64(TimeUnit::Nanosecond) => {
209	make_time!(array::Time64NanosecondArray, column, row)	×
210	}
211	DataType::Duration(TimeUnit::Microsecond) => {	×
212	make_duration!(array::DurationMicrosecondArray, column, row)	×
213	}
214	DataType::Duration(TimeUnit::Millisecond) => {	×
215	make_duration!(array::DurationMillisecondArray, column, row)	×
216	}
217	DataType::Duration(TimeUnit::Nanosecond) => {	×
218	make_duration!(array::DurationNanosecondArray, column, row)	×
219	}
220	DataType::Duration(TimeUnit::Second) => {	×
221	make_duration!(array::DurationSecondArray, column, row)	×
222	}
223	DataType::Binary => make_binary!(array::BinaryArray, column, row),	×
224	DataType::FixedSizeBinary(_) => make_binary!(array::FixedSizeBinaryArray, column, row),	×
225	DataType::LargeBinary => make_binary!(array::LargeBinaryArray, column, row),	×
226	DataType::Utf8 => make_from!(array::StringArray, column, row),	×
227	DataType::LargeUtf8 => make_from!(array::StringArray, column, row),	×
228	// DataType::Interval(TimeUnit::) => make_from!(array::BooleanArray, x, x0),	×
229	// DataType::List(_) => {}	×
230	// DataType::FixedSizeList(_, _) => {}	×
231	// DataType::LargeList(_) => {}	×
232	// DataType::Struct(_) => {}	×
233	// DataType::Union(_, _, _) => {}
234	// DataType::Dictionary(_, _) => {}
235	// DataType::Decimal128(_, _) => {}
236	// DataType::Decimal256(_, _) => {}
237	// DataType::Map(_, _) => {}
238	_ => Err(FieldTypeNotSupported(column_name.to_string())),	×
239	}
240	}	×

getdozer / dozer / 4195006582

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