micromegas_analytics/lakehouse/
materialized_view.rs

1use super::{
2    audience_guard::AudienceGuard, lakehouse_context::LakehouseContext,
3    partition_cache::QueryPartitionProvider,
4    partitioned_execution_plan::make_partitioned_execution_plan, reader_factory::ReaderFactory,
5    view::View,
6};
7use crate::time::TimeRange;
8use async_trait::async_trait;
9use datafusion::{
10    arrow::datatypes::SchemaRef,
11    catalog::{Session, TableProvider},
12    datasource::TableType,
13    error::DataFusionError,
14    logical_expr::{Expr, TableProviderFilterPushDown},
15    physical_plan::ExecutionPlan,
16};
17use micromegas_ingestion::data_lake_connection::is_read_only_violation;
18use micromegas_tracing::prelude::*;
19use std::sync::Arc;
20
21/// A DataFusion `TableProvider` for materialized views.
22#[derive(Debug)]
23pub struct MaterializedView {
24    lakehouse: Arc<LakehouseContext>,
25    reader_factory: Arc<ReaderFactory>,
26    view: Arc<dyn View>,
27    part_provider: Arc<dyn QueryPartitionProvider>,
28    query_range: Option<TimeRange>,
29    /// The `view_instance(...)` scan-time audience check, run before `jit_update`.
30    /// `Some` only when this provider was built by `ViewInstanceTableFunction` -- a
31    /// caller-named instance. `None` for every server-constructed `MaterializedView`: the
32    /// implicitly-registered global tables (`query.rs::register_table`) and `OwnershipRewrite`'s
33    /// own `processes`/`streams` sources, which are the row-level filter's job to filter row-by-row and must
34    /// never be denied wholesale.
35    instance_guard: Option<Arc<AudienceGuard>>,
36}
37
38impl MaterializedView {
39    pub fn new(
40        lakehouse: Arc<LakehouseContext>,
41        reader_factory: Arc<ReaderFactory>,
42        view: Arc<dyn View>,
43        part_provider: Arc<dyn QueryPartitionProvider>,
44        query_range: Option<TimeRange>,
45        instance_guard: Option<Arc<AudienceGuard>>,
46    ) -> Self {
47        Self {
48            lakehouse,
49            reader_factory,
50            view,
51            part_provider,
52            query_range,
53            instance_guard,
54        }
55    }
56
57    pub fn get_view(&self) -> Arc<dyn View> {
58        self.view.clone()
59    }
60
61    /// Whether `AudienceGuard::authorize_view_instance` will resolve this view's instance id
62    /// against the caller's scope, and deny, before `scan` yields a row. True only for a
63    /// caller-named, non-`'global'` `view_instance(...)`: those take the guard's Uuid arm (or its
64    /// fail-closed fallthrough). `'global'` is excluded because the guard passes it unconditionally
65    /// -- global instances are row-filtered instead.
66    ///
67    /// Deliberately does not mirror the guard's public-view-set arm: this accessor would still
68    /// return `true` for a public view set's instance, even though the guard passes it
69    /// unconditionally before ever reaching the Uuid arm. Harmless --
70    /// `OwnershipRewrite::predicate_for`'s own public-view-set check already returns `Ok(None)`
71    /// before this accessor is ever consulted, and both checks read
72    /// `IsolationConfig::public_view_sets`.
73    ///
74    /// Kept in step with `AudienceGuard::authorize_view_instance`'s arms; changing those means
75    /// revisiting this.
76    pub fn instance_is_audience_guarded(&self) -> bool {
77        self.instance_guard.is_some() && self.view.get_view_instance_id().as_str() != "global"
78    }
79}
80
81#[async_trait]
82impl TableProvider for MaterializedView {
83    fn schema(&self) -> SchemaRef {
84        self.view.get_file_schema()
85    }
86
87    fn table_type(&self) -> TableType {
88        TableType::Base
89    }
90
91    #[span_fn]
92    async fn scan(
93        &self,
94        state: &dyn Session,
95        projection: Option<&Vec<usize>>,
96        filters: &[Expr],
97        limit: Option<usize>,
98    ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
99        if let Some(guard) = &self.instance_guard {
100            guard
101                .authorize_view_instance(
102                    &self.view.get_view_set_name(),
103                    &self.view.get_view_instance_id(),
104                )
105                .await?;
106        }
107
108        if let Err(e) = self
109            .view
110            .jit_update(self.lakehouse.clone(), self.query_range)
111            .await
112        {
113            if is_read_only_violation(&e) {
114                warn!(
115                    "jit_update failed on a read-only connection for {}/{}: {e:#}",
116                    self.view.get_view_set_name(),
117                    self.view.get_view_instance_id()
118                );
119                imetric!("jit_update_failed_read_only", "count", 1_u64);
120                return Err(DataFusionError::External(
121                    format!(
122                        "the lakehouse is on a read-only connection and the requested range needs \
123                         partitions that aren't materialized yet for {}/{}",
124                        self.view.get_view_set_name(),
125                        self.view.get_view_instance_id()
126                    )
127                    .into(),
128                ));
129            }
130            return Err(DataFusionError::External(format!("{e:#}").into()));
131        }
132
133        let partitions = self
134            .part_provider
135            .fetch(
136                &self.view.get_view_set_name(),
137                &self.view.get_view_instance_id(),
138                self.query_range,
139                self.view.get_file_schema_hash(),
140            )
141            .await
142            .map_err(|e| datafusion::error::DataFusionError::External(e.into()))?;
143        trace!("MaterializedView::scan nb_partitions={}", partitions.len());
144
145        make_partitioned_execution_plan(
146            self.schema(),
147            self.reader_factory.clone(),
148            state,
149            projection,
150            filters,
151            limit,
152            Arc::new(partitions),
153            &self.view.get_scan_output_ordering(),
154        )
155    }
156
157    /// Tell DataFusion to push filters down to the scan method
158    fn supports_filters_pushdown(
159        &self,
160        filters: &[&Expr],
161    ) -> datafusion::error::Result<Vec<TableProviderFilterPushDown>> {
162        // Inexact because the pruning can't handle all expressions and pruning
163        // is not done at the row level -- there may be rows in returned files
164        // that do not pass the filter
165        Ok(vec![TableProviderFilterPushDown::Inexact; filters.len()])
166    }
167}
micromegas_analytics/lakehouse/materialized_view.rs

micromegas_analytics/lakehouse/
materialized_view.rs
