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    partitioned_execution_plan::{OrderingBounds, ScanOrdering},
10    session_configurator::NoOpSessionConfigurator,
11    view::{PartitionSpec, ScanSortColumn, View, ViewMetadata},
12    view_factory::ViewFactory,
13};
14use crate::audience::{audience_column_mismatch, coalesced_audience_column};
15use crate::time::{TimeRange, datetime_to_scalar};
16use anyhow::{Context, Result};
17use async_trait::async_trait;
18use chrono::{DateTime, TimeDelta, Utc};
19use datafusion::{
20    arrow::datatypes::{DataType, Field, Schema, TimeUnit},
21    logical_expr::{Expr, col},
22    prelude::*,
23};
24use std::sync::Arc;
25
26const VIEW_SET_NAME: &str = "blocks";
27const VIEW_INSTANCE_ID: &str = "global";
28lazy_static::lazy_static! {
29    static ref BEGIN_TIME_COLUMN: Arc<String> = Arc::new( String::from("begin_time"));
30    static ref INSERT_TIME_COLUMN: Arc<String> = Arc::new( String::from("insert_time"));
31}
32
33/// The single sort guarantee this view ever records or declares -- see the plan's Trade-offs
34/// section for why `block_id` is deliberately not part of it.
35fn insert_time_sort_order() -> Vec<String> {
36    vec![String::from("insert_time")]
37}
38
39/// The NULL-tolerant audience-mismatch *keep* predicate: true for a block whose own `audience`
40/// stamp does not disagree with either row it joins to (a NULL on either side always passes
41/// through -- see `audience_column_mismatch`'s doc comment for why). Shared with
42/// `source_count_query` and the mismatch counter in `maintenance.rs` so all three stay in sync.
43fn audience_mismatch_keep_predicate() -> String {
44    format!(
45        "NOT ({} OR {})",
46        audience_column_mismatch("blocks", "streams"),
47        audience_column_mismatch("blocks", "processes"),
48    )
49}
50
51/// True when every partition in `partitions_to_merge` either contributes no rows or already
52/// carries the exact `sort_order` this view can trust: only then can a merge safely declare and
53/// record the `insert_time` guarantee.
54fn all_inputs_ordered_or_empty(partitions_to_merge: &[Partition]) -> bool {
55    let wanted = insert_time_sort_order();
56    partitions_to_merge
57        .iter()
58        .all(|p| p.is_empty() || p.sort_order.as_ref() == Some(&wanted))
59}
60
61/// A view of the `blocks` table, providing access to telemetry block metadata.
62#[derive(Debug)]
63pub struct BlocksView {
64    view_set_name: Arc<String>,
65    view_instance_id: Arc<String>,
66    data_sql: Arc<String>,
67    /// Bound as `data_sql`'s `$3` -- the audience a never-stamped block (a NULL `blocks.audience`,
68    /// i.e. a legacy pre-v8 row) materializes under. Carried from
69    /// `LakehouseContext::default_audience` by every constructor call.
70    default_audience: Arc<str>,
71    ordered_merger: Arc<dyn PartitionMerger>,
72    plain_merger: Arc<dyn PartitionMerger>,
73}
74
75impl BlocksView {
76    /// `default_audience` is the deployment's `MICROMEGAS_DEFAULT_AUDIENCE`, sourced from
77    /// `LakehouseContext::default_audience`. It is one of three read sites, and the
78    /// only one whose resolved value is *baked into a partition*: every row of one materialization
79    /// sees the same bound value, so a never-stamped process is labelled consistently within a
80    /// partition. Two partitions materialized under different configured defaults can disagree --
81    /// changing the default requires regenerating the six views over the affected range.
82    pub fn new(default_audience: Arc<str>) -> Result<Self> {
83        let keep_predicate = audience_mismatch_keep_predicate();
84        let data_sql = Arc::new(format!(
85            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,
86           streams.dependencies_metadata, streams.objects_metadata, streams.tags, streams.properties, streams.insert_time as stream_insert_time, streams.format,
87           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,
88           {audience_column} AS audience
89         FROM blocks, streams, processes
90         WHERE blocks.stream_id = streams.stream_id
91         AND blocks.process_id = processes.process_id
92         AND blocks.insert_time >= $1
93         AND blocks.insert_time < $2
94         AND {keep_predicate}
95         ORDER BY blocks.insert_time, blocks.block_id
96         ;"#,
97            audience_column = coalesced_audience_column("blocks", 3),
98        ));
99        let empty_view_factory = Arc::new(ViewFactory::new(vec![]));
100        let schema = Arc::new(blocks_view_schema());
101        let ordered_merger: Arc<dyn PartitionMerger> = Arc::new(
102            QueryMerger::new(
103                empty_view_factory.clone(),
104                Arc::new(NoOpSessionConfigurator),
105                schema.clone(),
106                Arc::new(String::from("SELECT * FROM source ORDER BY insert_time;")),
107            )
108            .with_merge_scan_ordering(ScanOrdering::Concatenated {
109                columns: vec![ScanSortColumn {
110                    column: Arc::new(String::from("insert_time")),
111                    descending: false,
112                }],
113                bounds: OrderingBounds::InsertTime,
114            }),
115        );
116        let plain_merger: Arc<dyn PartitionMerger> = Arc::new(QueryMerger::new(
117            empty_view_factory,
118            Arc::new(NoOpSessionConfigurator),
119            schema,
120            Arc::new(String::from("SELECT * FROM source;")),
121        ));
122        Ok(Self {
123            view_set_name: Arc::new(String::from(VIEW_SET_NAME)),
124            view_instance_id: Arc::new(String::from(VIEW_INSTANCE_ID)),
125            data_sql,
126            default_audience,
127            ordered_merger,
128            plain_merger,
129        })
130    }
131}
132
133#[async_trait]
134impl View for BlocksView {
135    fn get_view_set_name(&self) -> Arc<String> {
136        self.view_set_name.clone()
137    }
138
139    fn get_view_instance_id(&self) -> Arc<String> {
140        self.view_instance_id.clone()
141    }
142
143    async fn make_batch_partition_spec(
144        &self,
145        lakehouse: Arc<LakehouseContext>,
146        _existing_partitions: Arc<PartitionCache>,
147        insert_range: TimeRange,
148    ) -> Result<Arc<dyn PartitionSpec>> {
149        let view_meta = ViewMetadata {
150            view_set_name: self.get_view_set_name(),
151            view_instance_id: self.get_view_instance_id(),
152            file_schema_hash: self.get_file_schema_hash(),
153        };
154        // `count` is the kept side of the audience-mismatch comparison (this partition's
155        // `record_count`); `unfiltered` is the same join and range with no predicate applied.
156        // One query gives `fetch_metadata_partition_spec` both, avoiding a second round trip.
157        let source_count_query = format!(
158            "SELECT COUNT(*) FILTER (WHERE {keep_predicate}) AS count,
159                    COUNT(*) AS unfiltered
160             FROM blocks, streams, processes
161             WHERE blocks.stream_id = streams.stream_id
162             AND blocks.process_id = processes.process_id
163             AND blocks.insert_time >= $1
164             AND blocks.insert_time < $2
165             ;",
166            keep_predicate = audience_mismatch_keep_predicate(),
167        );
168        let spec = fetch_metadata_partition_spec(
169            &lakehouse.lake().db_pool,
170            &source_count_query,
171            self.data_sql.clone(),
172            view_meta,
173            self.get_file_schema(),
174            insert_range,
175            self.get_time_bounds(),
176            Some(insert_time_sort_order()),
177            Some(self.default_audience.clone()),
178        )
179        .await
180        .with_context(|| "fetch_metadata_partition_spec")?;
181        Ok(Arc::new(spec))
182    }
183
184    fn get_file_schema_hash(&self) -> Vec<u8> {
185        blocks_file_schema_hash()
186    }
187
188    fn get_file_schema(&self) -> Arc<Schema> {
189        Arc::new(blocks_view_schema())
190    }
191
192    async fn jit_update(
193        &self,
194        _lakehouse: Arc<LakehouseContext>,
195        _query_range: Option<TimeRange>,
196    ) -> Result<()> {
197        if *self.view_instance_id == "global" {
198            // this view instance is updated using the deamon
199            return Ok(());
200        }
201        anyhow::bail!("not supported");
202    }
203
204    fn make_time_filter(&self, begin: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Expr>> {
205        Ok(vec![
206            col("begin_time").lt_eq(lit(datetime_to_scalar(end))),
207            col("insert_time").gt_eq(lit(datetime_to_scalar(begin))),
208        ])
209    }
210
211    fn get_time_bounds(&self) -> Arc<dyn DataFrameTimeBounds> {
212        //todo: make more robust, by changing to [ min(begin, insert), max(end, insert) ]
213        Arc::new(NamedColumnsTimeBounds::new(
214            BEGIN_TIME_COLUMN.clone(),
215            INSERT_TIME_COLUMN.clone(),
216        ))
217    }
218
219    fn get_update_group(&self) -> Option<i32> {
220        Some(1000)
221    }
222
223    fn get_max_partition_time_delta(&self, strategy: &PartitionCreationStrategy) -> TimeDelta {
224        match strategy {
225            PartitionCreationStrategy::Abort | PartitionCreationStrategy::CreateFromSource => {
226                TimeDelta::hours(1)
227            }
228            PartitionCreationStrategy::MergeExisting(_partitions) => TimeDelta::days(1),
229        }
230    }
231
232    async fn merge_partitions(
233        &self,
234        lakehouse: Arc<LakehouseContext>,
235        partitions_to_merge: Arc<Vec<Partition>>,
236        partitions_all_views: Arc<PartitionCache>,
237        insert_range: TimeRange,
238    ) -> Result<MergeQueryResult> {
239        // An all-empty source scans as an EmptyExec, whose SortExec is never elided -- taking the
240        // ordered path there would trip the plan-shape check's memory-regression warning on every
241        // quiet-day retry. So the ordered path additionally requires at least one non-empty input.
242        let any_non_empty = partitions_to_merge.iter().any(|p| !p.is_empty());
243        let merger = if any_non_empty && all_inputs_ordered_or_empty(&partitions_to_merge) {
244            &self.ordered_merger
245        } else {
246            &self.plain_merger
247        };
248        merger
249            .execute_merge_query(
250                lakehouse,
251                partitions_to_merge,
252                partitions_all_views,
253                insert_range,
254            )
255            .await
256    }
257
258    fn get_merged_partition_sort_order(
259        &self,
260        partitions_to_merge: &[Partition],
261    ) -> Option<Vec<String>> {
262        if all_inputs_ordered_or_empty(partitions_to_merge) {
263            Some(insert_time_sort_order())
264        } else {
265            None
266        }
267    }
268}
269
270/// Returns the Arrow schema for the blocks view.
271pub fn blocks_view_schema() -> Schema {
272    Schema::new(vec![
273        Field::new("block_id", DataType::Utf8, false),
274        Field::new("stream_id", DataType::Utf8, false),
275        Field::new("process_id", DataType::Utf8, false),
276        Field::new(
277            "begin_time",
278            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
279            false,
280        ),
281        Field::new("begin_ticks", DataType::Int64, false),
282        Field::new(
283            "end_time",
284            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
285            false,
286        ),
287        Field::new("end_ticks", DataType::Int64, false),
288        Field::new("nb_objects", DataType::Int32, false),
289        Field::new("object_offset", DataType::Int64, false),
290        Field::new("payload_size", DataType::Int64, false),
291        Field::new(
292            "insert_time",
293            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
294            false,
295        ),
296        Field::new("streams.dependencies_metadata", DataType::Binary, false),
297        Field::new("streams.objects_metadata", DataType::Binary, false),
298        Field::new(
299            "streams.tags",
300            DataType::List(Arc::new(Field::new("tag", DataType::Utf8, false))),
301            true,
302        ),
303        Field::new(
304            "streams.properties",
305            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Binary)),
306            false,
307        ),
308        Field::new(
309            "streams.insert_time",
310            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
311            false,
312        ),
313        Field::new("streams.format", DataType::Utf8, false),
314        Field::new(
315            "processes.start_time",
316            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
317            false,
318        ),
319        Field::new("processes.start_ticks", DataType::Int64, false),
320        Field::new("processes.tsc_frequency", DataType::Int64, false),
321        Field::new("processes.exe", DataType::Utf8, false),
322        Field::new("processes.username", DataType::Utf8, false),
323        Field::new("processes.realname", DataType::Utf8, false),
324        Field::new("processes.computer", DataType::Utf8, false),
325        Field::new("processes.distro", DataType::Utf8, false),
326        Field::new("processes.cpu_brand", DataType::Utf8, false),
327        Field::new(
328            "processes.insert_time",
329            DataType::Timestamp(TimeUnit::Nanosecond, Some("+00:00".into())),
330            false,
331        ),
332        // Nullable, not because the Postgres column is (no `processes`/`streams`/`blocks` column
333        // is `NOT NULL`) but because it is `NULL` in practice for every OTLP process and every
334        // root native process (`parent_process_id: Option<Uuid>`). The write path's nullability
335        // guard (`write_partition.rs`) checks declared-non-nullable columns against the batch, so
336        // a wrongly-`false` declaration here would reject essentially every fresh partition.
337        Field::new("processes.parent_process_id", DataType::Utf8, true),
338        Field::new(
339            "processes.properties",
340            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Binary)),
341            false,
342        ),
343        // Appended last: the block's own audience stamp, never derived from the `process_id`/
344        // `stream_id` it joins to. Non-nullable because `data_sql` wraps the extraction in
345        // `COALESCE(blocks.audience, $3)`, so a never-stamped (legacy pre-v8) block materializes
346        // under the deployment default instead of NULL; a block whose stamp disagrees with the
347        // `streams`/`processes` row it joins to is excluded from materialization entirely by
348        // `data_sql`'s mismatch predicate. Dictionary-encoded like every other view's audience
349        // column; `sql_arrow_bridge` delivers it as plain `Utf8`, so
350        // `metadata_partition_spec::cast_to_file_schema` casts it to this declared type before
351        // the write.
352        Field::new(
353            "audience",
354            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
355            false,
356        ),
357    ])
358}
359
360/// Returns the file schema hash for the blocks view.
361///
362/// Stays `vec![5]` -- deliberately not bumped, since the Arrow schema is unchanged and existing
363/// partitions remain valid to read under it. An older partition may contain a row the new
364/// audience-mismatch predicate would now exclude (all pre-existing lake data is public, so this
365/// is a consistency gap, not a confidentiality one); `regenerate_partitions` over `blocks` lets
366/// an operator get uniform, per-row semantics sooner than the retention window would.
367pub fn blocks_file_schema_hash() -> Vec<u8> {
368    vec![5] // Bumped from vec![3] for the dictionary-encoded `audience` column
369}
micromegas_analytics/lakehouse/blocks_view.rs

micromegas_analytics/lakehouse/
blocks_view.rs
