micromegas_analytics/lakehouse/
streams_view.rs

1use super::{
2    session_configurator::NoOpSessionConfigurator, sql_batch_view::SqlBatchView,
3    view_factory::ViewFactory,
4};
5use anyhow::Result;
6use chrono::TimeDelta;
7use datafusion::execution::runtime_env::RuntimeEnv;
8use micromegas_ingestion::data_lake_connection::DataLakeConnection;
9use std::sync::Arc;
10
11/// Creates a new `SqlBatchView` for streams.
12pub async fn make_streams_view(
13    runtime: Arc<RuntimeEnv>,
14    lake: Arc<DataLakeConnection>,
15    view_factory: Arc<ViewFactory>,
16) -> Result<SqlBatchView> {
17    let count_src_query = Arc::new(String::from(
18        r#"
19        SELECT count(*) as count
20        FROM  blocks
21        WHERE insert_time >= '{begin}'
22        AND   insert_time < '{end}'
23        ;"#,
24    ));
25    let transform_query = Arc::new(String::from(
26        r#"
27SELECT stream_id,
28       first_value("process_id") as process_id,
29       first_value("streams.dependencies_metadata") as dependencies_metadata,
30       first_value("streams.objects_metadata") as objects_metadata,
31       first_value("streams.tags") as tags,
32       first_value("streams.properties") as properties,
33       first_value("streams.insert_time") as insert_time,
34       first_value("streams.format") as format,
35       max(insert_time) as last_update_time,
36       arrow_cast(max(audience), 'Dictionary(Int32, Utf8)') as audience
37FROM blocks
38GROUP BY stream_id
39        ;"#,
40    ));
41    let merge_query = Arc::new(String::from(
42        r#"
43SELECT stream_id,
44       first_value(process_id) as process_id,
45       first_value(dependencies_metadata) as dependencies_metadata,
46       first_value(objects_metadata) as objects_metadata,
47       first_value(tags) as tags,
48       first_value(properties) as properties,
49       first_value(insert_time) as insert_time,
50       first_value(format) as format,
51       max(last_update_time) as last_update_time,
52       arrow_cast(max(audience), 'Dictionary(Int32, Utf8)') as audience
53FROM {source}
54GROUP BY stream_id
55        ;"#,
56    ));
57    let min_time_column = Arc::new(String::from("insert_time"));
58    let max_time_column = Arc::new(String::from("last_update_time"));
59    SqlBatchView::new(
60        runtime,
61        Arc::new("streams".to_owned()),
62        min_time_column,
63        max_time_column,
64        count_src_query,
65        transform_query,
66        merge_query,
67        lake,
68        view_factory,
69        Arc::new(NoOpSessionConfigurator),
70        Some(2000),
71        TimeDelta::days(1), // from source
72        TimeDelta::days(1), // when merging
73    )
74    .await
75}
micromegas_analytics/lakehouse/streams_view.rs

micromegas_analytics/lakehouse/
streams_view.rs
