micromegas_analytics/lakehouse/
blocks_view.rs

1use super::{
2    batch_update::PartitionCreationStrategy,
3    dataframe_time_bounds::{DataFrameTimeBounds, NamedColumnsTimeBounds},
4    lakehouse_context::LakehouseContext,
5    merge::{MergeQueryResult, PartitionMerger, QueryMerger},
6    metadata_partition_spec::fetch_metadata_partition_spec,
7    partition::Partition,
8    partition_cache::PartitionCache,
9    session_configurator::NoOpSessionConfigurator,
10    view::{PartitionSpec, ScanSortColumn, View, ViewMetadata},
11    view_factory::ViewFactory,
12};
13use crate::time::{TimeRange, datetime_to_scalar};
14use anyhow::{Context, Result};
15use async_trait::async_trait;
16use chrono::{DateTime, TimeDelta, Utc};
17use datafusion::{
18    arrow::datatypes::{DataType, Field, Schema, TimeUnit},
19    logical_expr::{Expr, col},
20    prelude::*,
21};
22use std::sync::Arc;
23
24const VIEW_SET_NAME: &str = "blocks";
25const VIEW_INSTANCE_ID: &str = "global";
26lazy_static::lazy_static! {
27    static ref BEGIN_TIME_COLUMN: Arc<String> = Arc::new( String::from("begin_time"));
28    static ref INSERT_TIME_COLUMN: Arc<String> = Arc::new( String::from("insert_time"));
29}
30
31/// The single sort guarantee this view ever records or declares -- see the plan's Trade-offs
32/// section for why `block_id` is deliberately not part of it.
33fn insert_time_sort_order() -> Vec<String> {
34    vec![String::from("insert_time")]
35}
36
37/// True when every partition in `partitions_to_merge` either contributes no rows or already
38/// carries the exact `sort_order` this view can trust (Design §1/§4): only then can a merge
39/// safely declare and record the `insert_time` guarantee.
40fn all_inputs_ordered_or_empty(partitions_to_merge: &[Partition]) -> bool {
41    let wanted = insert_time_sort_order();
42    partitions_to_merge
43        .iter()
44        .all(|p| p.is_empty() || p.sort_order.as_ref() == Some(&wanted))
45}
46
47/// A view of the `blocks` table, providing access to telemetry block metadata.
48#[derive(Debug)]
49pub struct BlocksView {
50    view_set_name: Arc<String>,
51    view_instance_id: Arc<String>,
52    data_sql: Arc<String>,
53    ordered_merger: Arc<dyn PartitionMerger>,
54    plain_merger: Arc<dyn PartitionMerger>,
55}
56
57impl BlocksView {
58    pub fn new() -> Result<Self> {
59        let data_sql = Arc::new(String::from(
60            r#"SELECT block_id, streams.stream_id, processes.process_id, blocks.begin_time, blocks.begin_ticks, blocks.end_time, blocks.end_ticks, blocks.nb_objects, blocks.object_offset, blocks.payload_size, blocks.insert_time,
61           streams.dependencies_metadata, streams.objects_metadata, streams.tags, streams.properties, streams.insert_time as stream_insert_time, streams.format,
62           processes.start_time, processes.start_ticks, processes.tsc_frequency, processes.exe, processes.username, processes.realname, processes.computer, processes.distro, processes.cpu_brand, processes.insert_time as process_insert_time, processes.parent_process_id, processes.properties as process_properties
63         FROM blocks, streams, processes
64         WHERE blocks.stream_id = streams.stream_id
65         AND blocks.process_id = processes.process_id
66         AND blocks.insert_time >= $1
67         AND blocks.insert_time < $2
68         ORDER BY blocks.insert_time, blocks.block_id
69         ;"#,
70        ));
71        let empty_view_factory = Arc::new(ViewFactory::new(vec![]));
72        let schema = Arc::new(blocks_view_schema());
73        let ordered_merger: Arc<dyn PartitionMerger> = Arc::new(
74            QueryMerger::new(
75                empty_view_factory.clone(),
76                Arc::new(NoOpSessionConfigurator),
77                schema.clone(),
78                Arc::new(String::from("SELECT * FROM source ORDER BY insert_time;")),
79            )
80            .with_merge_scan_ordering(vec![ScanSortColumn {
81                column: Arc::new(String::from("insert_time")),
82                descending: false,
83            }]),
84        );
85        let plain_merger: Arc<dyn PartitionMerger> = Arc::new(QueryMerger::new(
86            empty_view_factory,
87            Arc::new(NoOpSessionConfigurator),
88            schema,
89            Arc::new(String::from("SELECT * FROM source;")),
90        ));
91        Ok(Self {
92            view_set_name: Arc::new(String::from(VIEW_SET_NAME)),
93            view_instance_id: Arc::new(String::from(VIEW_INSTANCE_ID)),
94            data_sql,
95            ordered_merger,
96            plain_merger,
97        })
98    }
99}
100
101#[async_trait]
102impl View for BlocksView {
103    fn get_view_set_name(&self) -> Arc<String> {
104        self.view_set_name.clone()
105    }
106
107    fn get_view_instance_id(&self) -> Arc<String> {
108        self.view_instance_id.clone()
109    }
110
111    async fn make_batch_partition_spec(
112        &self,
113        lakehouse: Arc<LakehouseContext>,
114        _existing_partitions: Arc<PartitionCache>,
115        insert_range: TimeRange,
116    ) -> Result<Arc<dyn PartitionSpec>> {
117        let view_meta = ViewMetadata {
118            view_set_name: self.get_view_set_name(),
119            view_instance_id: self.get_view_instance_id(),
120            file_schema_hash: self.get_file_schema_hash(),
121        };
122        let source_count_query = "
123             SELECT COUNT(*) as count
124             FROM blocks, streams, processes
125             WHERE blocks.stream_id = streams.stream_id
126             AND blocks.process_id = processes.process_id
127             AND blocks.insert_time >= $1
128             AND blocks.insert_time < $2
129             ;";
130        Ok(Arc::new(
131            fetch_metadata_partition_spec(
132                &lakehouse.lake().db_pool,
133                source_count_query,
134                self.data_sql.clone(),
135                view_meta,
136                self.get_file_schema(),
137                insert_range,
138                self.get_time_bounds(),
139                Some(insert_time_sort_order()),
140            )
141            .await
142            .with_context(|| "fetch_metadata_partition_spec")?,
143        ))
144    }
145
146    fn get_file_schema_hash(&self) -> Vec<u8> {
147        blocks_file_schema_hash()
148    }
149
150    fn get_file_schema(&self) -> Arc<Schema> {
151        Arc::new(blocks_view_schema())
152    }
153
154    async fn jit_update(
155        &self,
156        _lakehouse: Arc<LakehouseContext>,
157        _query_range: Option<TimeRange>,
158    ) -> Result<()> {
159        if *self.view_instance_id == "global" {
160            // this view instance is updated using the deamon
161            return Ok(());
162        }
163        anyhow::bail!("not supported");
164    }
165
166    fn make_time_filter(&self, begin: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Expr>> {
167        Ok(vec![
168            col("begin_time").lt_eq(lit(datetime_to_scalar(end))),
169            col("insert_time").gt_eq(lit(datetime_to_scalar(begin))),
170        ])
171    }
172
173    fn get_time_bounds(&self) -> Arc<dyn DataFrameTimeBounds> {
174        //todo: make more robust, by changing to [ min(begin, insert), max(end, insert) ]
175        Arc::new(NamedColumnsTimeBounds::new(
176            BEGIN_TIME_COLUMN.clone(),
177            INSERT_TIME_COLUMN.clone(),
178        ))
179    }
180
181    fn get_update_group(&self) -> Option<i32> {
182        Some(1000)
183    }
184
185    fn get_max_partition_time_delta(&self, strategy: &PartitionCreationStrategy) -> TimeDelta {
186        match strategy {
187            PartitionCreationStrategy::Abort | PartitionCreationStrategy::CreateFromSource => {
188                TimeDelta::hours(1)
189            }
190            PartitionCreationStrategy::MergeExisting(_partitions) => TimeDelta::days(1),
191        }
192    }
193
194    async fn merge_partitions(
195        &self,
196        lakehouse: Arc<LakehouseContext>,
197        partitions_to_merge: Arc<Vec<Partition>>,
198        partitions_all_views: Arc<PartitionCache>,
199        insert_range: TimeRange,
200    ) -> Result<MergeQueryResult> {
201        // An all-empty source scans as an EmptyExec, whose SortExec is never elided -- taking the
202        // ordered path there would trip the plan-shape check's memory-regression warning on every
203        // quiet-day retry. So the ordered path additionally requires at least one non-empty input.
204        let any_non_empty = partitions_to_merge.iter().any(|p| !p.is_empty());
205        let merger = if any_non_empty && all_inputs_ordered_or_empty(&partitions_to_merge) {
206            &self.ordered_merger
207        } else {
208            &self.plain_merger
209        };
210        merger
211            .execute_merge_query(
212                lakehouse,
213                partitions_to_merge,
214                partitions_all_views,
215                insert_range,
216            )
217            .await
218    }
219
220    fn get_merged_partition_sort_order(
221        &self,
222        partitions_to_merge: &[Partition],
223    ) -> Option<Vec<String>> {
224        if all_inputs_ordered_or_empty(partitions_to_merge) {
225            Some(insert_time_sort_order())
226        } else {
227            None
228        }
229    }
230}
231
232/// Returns the Arrow schema for the blocks view.
233pub fn blocks_view_schema() -> Schema {
234    Schema::new(vec![
235        Field::new("block_id", DataType::Utf8, false),
236        Field::new("stream_id", DataType::Utf8, false),
237        Field::new("process_id", DataType::Utf8, false),
238        Field::new(
239            "begin_time",
240            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
241            false,
242        ),
243        Field::new("begin_ticks", DataType::Int64, false),
244        Field::new(
245            "end_time",
246            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
247            false,
248        ),
249        Field::new("end_ticks", DataType::Int64, false),
250        Field::new("nb_objects", DataType::Int32, false),
251        Field::new("object_offset", DataType::Int64, false),
252        Field::new("payload_size", DataType::Int64, false),
253        Field::new(
254            "insert_time",
255            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
256            false,
257        ),
258        Field::new("streams.dependencies_metadata", DataType::Binary, false),
259        Field::new("streams.objects_metadata", DataType::Binary, false),
260        Field::new(
261            "streams.tags",
262            DataType::List(Arc::new(Field::new("tag", DataType::Utf8, false))),
263            true,
264        ),
265        Field::new(
266            "streams.properties",
267            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Binary)),
268            false,
269        ),
270        Field::new(
271            "streams.insert_time",
272            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
273            false,
274        ),
275        Field::new("streams.format", DataType::Utf8, false),
276        Field::new(
277            "processes.start_time",
278            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
279            false,
280        ),
281        Field::new("processes.start_ticks", DataType::Int64, false),
282        Field::new("processes.tsc_frequency", DataType::Int64, false),
283        Field::new("processes.exe", DataType::Utf8, false),
284        Field::new("processes.username", DataType::Utf8, false),
285        Field::new("processes.realname", DataType::Utf8, false),
286        Field::new("processes.computer", DataType::Utf8, false),
287        Field::new("processes.distro", DataType::Utf8, false),
288        Field::new("processes.cpu_brand", DataType::Utf8, false),
289        Field::new(
290            "processes.insert_time",
291            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
292            false,
293        ),
294        Field::new("processes.parent_process_id", DataType::Utf8, false),
295        Field::new(
296            "processes.properties",
297            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Binary)),
298            false,
299        ),
300    ])
301}
302
303/// Returns the file schema hash for the blocks view.
304pub fn blocks_file_schema_hash() -> Vec<u8> {
305    vec![3] // Bumped from vec![2] for streams.format column (OTLP support)
306}
micromegas_analytics/lakehouse/blocks_view.rs

micromegas_analytics/lakehouse/
blocks_view.rs
