4392484403

Build Type

push

github

Committed by GitHub

Commit Message

feat: Asynchoronous indexing (#1206)

Run Details

270 of 270 new or added lines in 13 files covered. (100.0%)

28714 of 38777 relevant lines covered (74.05%)

89484.24 hits per line

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

98.53

/dozer-api/src/test_utils.rs

use dozer_types::serde_json::{json, Value};
use dozer_types::types::{Field, Record, SchemaWithIndex, SourceDefinition};
use dozer_types::{
    models::api_endpoint::{ApiEndpoint, ApiIndex},
    types::{FieldDefinition, FieldType, IndexDefinition, Schema, SchemaIdentifier},
};

use dozer_cache::cache::{CacheManager, LmdbCacheManager, RecordWithId};

pub fn get_schema() -> SchemaWithIndex {
    let fields = vec![
        FieldDefinition {
            name: "film_id".to_string(),
            typ: FieldType::UInt,
            nullable: false,
            source: SourceDefinition::Dynamic,
        },
        FieldDefinition {
            name: "description".to_string(),
            typ: FieldType::String,
            nullable: true,
            source: SourceDefinition::Dynamic,
        },
        FieldDefinition {
            name: "rental_rate".to_string(),
            typ: FieldType::Float,
            nullable: true,
            source: SourceDefinition::Dynamic,
        },
        FieldDefinition {
            name: "release_year".to_string(),
            typ: FieldType::UInt,
            nullable: true,
            source: SourceDefinition::Dynamic,
        },
        FieldDefinition {
            name: "updated_at".to_string(),
            typ: FieldType::Timestamp,
            nullable: true,
            source: SourceDefinition::Dynamic,
        },
    ];
    let secondary_indexes = fields
        .iter()
        .enumerate()
        .map(|(idx, _f)| IndexDefinition::SortedInverted(vec![idx]))
        .collect();
    (
        Schema {
            identifier: Some(SchemaIdentifier {
                id: 3003108387,
                version: 1,
            }),
            fields,
            primary_index: vec![0],
        },
        secondary_indexes,
    )
}

pub fn get_endpoint() -> ApiEndpoint {
    ApiEndpoint {
        name: "films".to_string(),
        path: "/films".to_string(),
        index: Some(ApiIndex {
            primary_key: vec!["film_id".to_string()],
        }),
        table_name: "film".to_string(),
    }
}

fn get_films() -> Vec<Value> {
    let mut result = vec![
        json!({
          "description": "A Amazing Panorama of a Mad Scientist And a Husband who must Meet a Woman in The Outback",
          "rental_rate": null,
          "release_year": 2006,
          "film_id": 268,
          "updated_at": null
        }),
        json!({
          "film_id": 524,
          "release_year": 2006,
          "rental_rate": null,
          "description": "A Intrepid Display of a Pastry Chef And a Cat who must Kill a A Shark in Ancient China",
          "updated_at": null
        }),
    ];

    for film_id in 1..=50 {
        result.push(json!({
            "film_id": film_id,
            "description": format!("Film {film_id}"),
            "rental_rate": null,
            "release_year": 2006,
            "updated_at": null
        }));
    }
    result
}

pub fn initialize_cache(
    schema_name: &str,
    schema: Option<SchemaWithIndex>,
) -> Box<dyn CacheManager> {
    let cache_manager = LmdbCacheManager::new(Default::default()).unwrap();
    let (schema, secondary_indexes) = schema.unwrap_or_else(get_schema);
    let cache = cache_manager
        .create_cache(schema.clone(), secondary_indexes)
        .unwrap();
    let records = get_sample_records(schema);
    for mut record in records {
        cache.insert(&mut record.record).unwrap();
    }
    cache.commit().unwrap();
    cache_manager.wait_until_indexing_catchup();

    cache_manager
        .create_alias(cache.name(), schema_name)
        .unwrap();
    Box::new(cache_manager)
}

pub fn get_sample_records(schema: Schema) -> Vec<RecordWithId> {
    let records_value: Vec<Value> = get_films();
    let mut records = vec![];
    for (record_index, record_str) in records_value.into_iter().enumerate() {
        let film_id = record_str["film_id"].as_u64();
        let description = record_str["description"].as_str();
        let release_year = record_str["release_year"].as_u64();
        if let (Some(film_id), Some(description), Some(release_year)) =
            (film_id, description, release_year)
        {
            let record = Record::new(
                schema.identifier,
                vec![
                    Field::UInt(film_id),
                    Field::String(description.to_string()),
                    Field::Null,
                    Field::UInt(release_year),
                    Field::Null,
                ],
                Some(1),
            );
            records.push(RecordWithId::new(record_index as _, record));
        }
    }
    records
}

1	use dozer_types::serde_json::{json, Value};
2	use dozer_types::types::{Field, Record, SchemaWithIndex, SourceDefinition};
3	use dozer_types::{
4	models::api_endpoint::{ApiEndpoint, ApiIndex},
5	types::{FieldDefinition, FieldType, IndexDefinition, Schema, SchemaIdentifier},
6	};
7
8	use dozer_cache::cache::{CacheManager, LmdbCacheManager, RecordWithId};
9
10	pub fn get_schema() -> SchemaWithIndex {	23✔
11	let fields = vec![	23✔
12	FieldDefinition {	23✔
13	name: "film_id".to_string(),	23✔
14	typ: FieldType::UInt,	23✔
15	nullable: false,	23✔
16	source: SourceDefinition::Dynamic,	23✔
17	},	23✔
18	FieldDefinition {	23✔
19	name: "description".to_string(),	23✔
20	typ: FieldType::String,	23✔
21	nullable: true,	23✔
22	source: SourceDefinition::Dynamic,	23✔
23	},	23✔
24	FieldDefinition {	23✔
25	name: "rental_rate".to_string(),	23✔
26	typ: FieldType::Float,	23✔
27	nullable: true,	23✔
28	source: SourceDefinition::Dynamic,	23✔
29	},	23✔
30	FieldDefinition {	23✔
31	name: "release_year".to_string(),	23✔
32	typ: FieldType::UInt,	23✔
33	nullable: true,	23✔
34	source: SourceDefinition::Dynamic,	23✔
35	},	23✔
36	FieldDefinition {	23✔
37	name: "updated_at".to_string(),	23✔
38	typ: FieldType::Timestamp,	23✔
39	nullable: true,	23✔
40	source: SourceDefinition::Dynamic,	23✔
41	},	23✔
42	];	23✔
43	let secondary_indexes = fields	23✔
44	.iter()	23✔
45	.enumerate()	23✔
46	.map(\|(idx, _f)\| IndexDefinition::SortedInverted(vec![idx]))	119✔
47	.collect();	23✔
48	(	23✔
49	Schema {	23✔
50	identifier: Some(SchemaIdentifier {	23✔
51	id: 3003108387,	23✔
52	version: 1,	23✔
53	}),	23✔
54	fields,	23✔
55	primary_index: vec![0],	23✔
56	},	23✔
57	secondary_indexes,	23✔
58	)	23✔
59	}	23✔
60
61	pub fn get_endpoint() -> ApiEndpoint {	23✔
62	ApiEndpoint {	23✔
63	name: "films".to_string(),	23✔
64	path: "/films".to_string(),	23✔
65	index: Some(ApiIndex {	23✔
66	primary_key: vec!["film_id".to_string()],	23✔
67	}),	23✔
68	table_name: "film".to_string(),	23✔
69	}	23✔
70	}	23✔
71
72	fn get_films() -> Vec<Value> {	20✔
73	let mut result = vec![	20✔
74	json!({	20✔
75	"description": "A Amazing Panorama of a Mad Scientist And a Husband who must Meet a Woman in The Outback",	20✔
76	"rental_rate": null,	20✔
77	"release_year": 2006,	20✔
78	"film_id": 268,	20✔
79	"updated_at": null	20✔
80	}),	20✔
81	json!({	20✔
82	"film_id": 524,	20✔
83	"release_year": 2006,	20✔
84	"rental_rate": null,	20✔
85	"description": "A Intrepid Display of a Pastry Chef And a Cat who must Kill a A Shark in Ancient China",	20✔
86	"updated_at": null	20✔
87	}),	20✔
88	];	20✔
89
90	for film_id in 1..=50 {	1,020✔
91	result.push(json!({	1,000✔
92	"film_id": film_id,	1,000✔
93	"description": format!("Film {film_id}"),	1,000✔
94	"rental_rate": null,	1,000✔
95	"release_year": 2006,	1,000✔
96	"updated_at": null	1,000✔
97	}));	1,000✔
98	}	1,000✔
99	result	20✔
100	}	20✔
101
102	pub fn initialize_cache(	19✔
103	schema_name: &str,	19✔
104	schema: Option<SchemaWithIndex>,	19✔
105	) -> Box<dyn CacheManager> {	19✔
106	let cache_manager = LmdbCacheManager::new(Default::default()).unwrap();	19✔
107	let (schema, secondary_indexes) = schema.unwrap_or_else(get_schema);	19✔
108	let cache = cache_manager	19✔
109	.create_cache(schema.clone(), secondary_indexes)	19✔
110	.unwrap();	19✔
111	let records = get_sample_records(schema);	19✔
112	for mut record in records {	1,007✔
113	cache.insert(&mut record.record).unwrap();	988✔
114	}	988✔
115	cache.commit().unwrap();	19✔
116	cache_manager.wait_until_indexing_catchup();	19✔
117		19✔
118	cache_manager	19✔
119	.create_alias(cache.name(), schema_name)	19✔
120	.unwrap();	19✔
121	Box::new(cache_manager)	19✔
122	}	19✔
123		×
124	pub fn get_sample_records(schema: Schema) -> Vec<RecordWithId> {	20✔
125	let records_value: Vec<Value> = get_films();	20✔
126	let mut records = vec![];	20✔
127	for (record_index, record_str) in records_value.into_iter().enumerate() {	1,040✔
128	let film_id = record_str["film_id"].as_u64();	1,040✔
129	let description = record_str["description"].as_str();	1,040✔
130	let release_year = record_str["release_year"].as_u64();	1,040✔
131	if let (Some(film_id), Some(description), Some(release_year)) =	1,040✔
132	(film_id, description, release_year)	1,040✔
133	{	1,040✔
134	let record = Record::new(	1,040✔
135	schema.identifier,	1,040✔
136	vec![	1,040✔
137	Field::UInt(film_id),	1,040✔
138	Field::String(description.to_string()),	1,040✔
139	Field::Null,	1,040✔
140	Field::UInt(release_year),	1,040✔
141	Field::Null,	1,040✔
142	],	1,040✔
143	Some(1),	1,040✔
144	);	1,040✔
145	records.push(RecordWithId::new(record_index as _, record));	1,040✔
146	}	1,040✔
147	}	×
148	records	20✔
149	}	20✔

getdozer / dozer / 4392484403

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