micromegas_analytics/lakehouse/
log_view.rs

1use super::{
2    batch_update::PartitionCreationStrategy,
3    block_partition_spec::{BlockPartitionSpec, BlockProcessor, BlockProcessorMap},
4    blocks_view::BlocksView,
5    dataframe_time_bounds::{DataFrameTimeBounds, NamedColumnsTimeBounds},
6    jit_partitions::{JitPartitionConfig, write_partition_from_blocks},
7    lakehouse_context::LakehouseContext,
8    log_block_processor::LogBlockProcessor,
9    otel::logs_block_processor::OtelLogsBlockProcessor,
10    partition_cache::PartitionCache,
11    partition_source_data::fetch_partition_source_data,
12    view::{PartitionSpec, View, ViewMetadata},
13    view_factory::ViewMaker,
14};
15use crate::{
16    lakehouse::jit_partitions::{
17        BlockOrder, find_up_to_date_partitions, generate_process_jit_partitions,
18    },
19    log_entries_table::log_table_schema,
20    metadata::find_process,
21    time::{TimeRange, datetime_to_scalar},
22};
23use anyhow::{Context, Result};
24use async_trait::async_trait;
25use chrono::{DateTime, TimeDelta, Utc};
26use datafusion::{
27    arrow::datatypes::Schema,
28    logical_expr::{Between, Expr, col},
29};
30use micromegas_ingestion::web_ingestion_service::{FORMAT_OTLP_LOGS, FORMAT_TRANSIT};
31use micromegas_tracing::prelude::*;
32use std::collections::HashMap;
33use std::sync::Arc;
34use uuid::Uuid;
35
36const VIEW_SET_NAME: &str = "log_entries";
37const SCHEMA_VERSION: u8 = 7;
38lazy_static::lazy_static! {
39    static ref TIME_COLUMN: Arc<String> = Arc::new( String::from("time"));
40}
41
42/// Block-processor map covering native transit + OTel log blocks.
43fn log_processors() -> Arc<BlockProcessorMap> {
44    let mut m: BlockProcessorMap = HashMap::new();
45    m.insert(
46        FORMAT_TRANSIT,
47        Arc::new(LogBlockProcessor {}) as Arc<dyn BlockProcessor>,
48    );
49    m.insert(
50        FORMAT_OTLP_LOGS,
51        Arc::new(OtelLogsBlockProcessor {}) as Arc<dyn BlockProcessor>,
52    );
53    Arc::new(m)
54}
55
56/// A `ViewMaker` for creating `LogView` instances.
57#[derive(Debug)]
58pub struct LogViewMaker {}
59
60impl ViewMaker for LogViewMaker {
61    fn make_view(&self, view_instance_id: &str) -> Result<Arc<dyn View>> {
62        Ok(Arc::new(LogView::new(view_instance_id)?))
63    }
64
65    fn get_schema_hash(&self) -> Vec<u8> {
66        vec![SCHEMA_VERSION]
67    }
68
69    fn get_schema(&self) -> Arc<Schema> {
70        Arc::new(log_table_schema())
71    }
72}
73
74/// A view of log entries.
75#[derive(Debug)]
76pub struct LogView {
77    view_set_name: Arc<String>,
78    view_instance_id: Arc<String>,
79    process_id: Option<sqlx::types::Uuid>,
80}
81
82impl LogView {
83    pub fn new(view_instance_id: &str) -> Result<Self> {
84        let process_id = if view_instance_id == "global" {
85            None
86        } else {
87            Some(Uuid::parse_str(view_instance_id).with_context(|| "Uuid::parse_str")?)
88        };
89
90        Ok(Self {
91            view_set_name: Arc::new(String::from(VIEW_SET_NAME)),
92            view_instance_id: Arc::new(view_instance_id.into()),
93            process_id,
94        })
95    }
96}
97
98#[async_trait]
99impl View for LogView {
100    fn get_view_set_name(&self) -> Arc<String> {
101        self.view_set_name.clone()
102    }
103
104    fn get_view_instance_id(&self) -> Arc<String> {
105        self.view_instance_id.clone()
106    }
107
108    async fn make_batch_partition_spec(
109        &self,
110        lakehouse: Arc<LakehouseContext>,
111        existing_partitions: Arc<PartitionCache>,
112        insert_range: TimeRange,
113    ) -> Result<Arc<dyn PartitionSpec>> {
114        if *self.view_instance_id != "global" {
115            anyhow::bail!("not supported for jit queries... should it?");
116        }
117        let source_data = Arc::new(
118            fetch_partition_source_data(
119                lakehouse.clone(),
120                existing_partitions,
121                insert_range,
122                "log",
123            )
124            .await
125            .with_context(|| "fetch_partition_source_data")?,
126        );
127        Ok(Arc::new(BlockPartitionSpec {
128            view_metadata: ViewMetadata {
129                view_set_name: self.view_set_name.clone(),
130                view_instance_id: self.view_instance_id.clone(),
131                file_schema_hash: self.get_file_schema_hash(),
132            },
133            schema: self.get_file_schema(),
134            insert_range,
135            source_data,
136            block_processors: log_processors(),
137        }))
138    }
139
140    fn get_file_schema_hash(&self) -> Vec<u8> {
141        vec![SCHEMA_VERSION]
142    }
143
144    fn get_file_schema(&self) -> Arc<Schema> {
145        Arc::new(log_table_schema())
146    }
147
148    #[span_fn]
149    async fn jit_update(
150        &self,
151        lakehouse: Arc<LakehouseContext>,
152        query_range: Option<TimeRange>,
153    ) -> Result<()> {
154        if *self.view_instance_id == "global" {
155            // this view instance is updated using the deamon
156            return Ok(());
157        }
158        let process = Arc::new(
159            find_process(
160                &lakehouse.lake().db_pool,
161                &self
162                    .process_id
163                    .with_context(|| "getting a view's process_id")?,
164                &lakehouse.default_audience(),
165            )
166            .await
167            .with_context(|| "find_process")?,
168        );
169        let query_range =
170            query_range.unwrap_or_else(|| TimeRange::new(process.start_time, chrono::Utc::now()));
171
172        let blocks_view = BlocksView::new(lakehouse.default_audience())?;
173        let all_partitions = generate_process_jit_partitions(
174            &JitPartitionConfig::default(),
175            lakehouse.clone(),
176            &blocks_view,
177            &query_range,
178            process.clone(),
179            "log",
180        )
181        .await
182        .with_context(|| "generate_process_jit_partitions")?;
183        let view_meta = ViewMetadata {
184            view_set_name: self.get_view_set_name(),
185            view_instance_id: self.get_view_instance_id(),
186            file_schema_hash: self.get_file_schema_hash(),
187        };
188        let block_processors = log_processors();
189
190        let up_to_date = find_up_to_date_partitions(
191            &lakehouse.lake().db_pool,
192            view_meta.clone(),
193            BlockOrder::InsertTime,
194            &all_partitions,
195        )
196        .await
197        .with_context(|| "find_up_to_date_partitions")?;
198        for (part, up_to_date) in all_partitions.into_iter().zip(up_to_date) {
199            if !up_to_date {
200                write_partition_from_blocks(
201                    lakehouse.lake().clone(),
202                    view_meta.clone(),
203                    self.get_file_schema(),
204                    part,
205                    block_processors.clone(),
206                )
207                .await
208                .with_context(|| "write_partition_from_blocks")?;
209            }
210        }
211        Ok(())
212    }
213
214    fn make_time_filter(&self, begin: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Expr>> {
215        Ok(vec![Expr::Between(Between::new(
216            col("time").into(),
217            false,
218            Expr::Literal(datetime_to_scalar(begin), None).into(),
219            Expr::Literal(datetime_to_scalar(end), None).into(),
220        ))])
221    }
222
223    fn get_time_bounds(&self) -> Arc<dyn DataFrameTimeBounds> {
224        Arc::new(NamedColumnsTimeBounds::new(
225            TIME_COLUMN.clone(),
226            TIME_COLUMN.clone(),
227        ))
228    }
229
230    fn get_update_group(&self) -> Option<i32> {
231        if *(self.get_view_instance_id()) == "global" {
232            Some(2000)
233        } else {
234            None
235        }
236    }
237
238    fn get_max_partition_time_delta(&self, _strategy: &PartitionCreationStrategy) -> TimeDelta {
239        TimeDelta::hours(1)
240    }
241}
micromegas_analytics/lakehouse/log_view.rs

micromegas_analytics/lakehouse/
log_view.rs
