micromegas_analytics/lakehouse/

ownership_rewrite.rs

//! `OwnershipRewrite` -- the row-level filter.
//!
//! An `AnalyzerRule` that injects an audience-filtering predicate into every
//! `MaterializedView`-backed `TableScan` a query plan touches, based on the caller's `ReadScope`
//! (`read_scope.rs`) and [`crate::lakehouse::read_scope::IsolationConfig`]'s
//! `MICROMEGAS_PUBLIC_VIEW_SETS`.
//!
//! ## `ReadScope::All` is a true no-op
//!
//! `ReadScope::All` is the internal/maintenance marker -- never the output of a real `ReadPolicy`
//! (see `read_scope.rs`). `analyze()` returns the plan unchanged when constructed with it, and
//! `query.rs::make_session_context` does not even construct this rule for `ReadScope::All`
//! callers, so their `ViewFactory` need not carry `processes`/`streams` either.
//!
//! ## The physical `audience` column
//!
//! `blocks`, `processes`, `streams`, `log_entries`, `measures`, and `log_stats` each carry a
//! physical `audience` column -- declared non-nullable on `blocks`/`log_entries`/`measures`, and
//! non-null by construction (though declared nullable in their inferred `SqlBatchView` schema) on
//! `processes`/`streams`/`log_stats` -- extracted once from Postgres's own `audience` column at
//! the `blocks` view's materialization (its own row's stamp) and propagated structurally into
//! every downstream view. A DDL-defined materialized view carrying its own `audience` column
//! (enforced at `CREATE` time, see `view_definition.rs`) joins this group too. Every view
//! carrying an `audience` column is consequently filtered with a bare `Filter` on that column
//! (the column-filter branch in the table below): no semi-join, no `property_get`, no
//! per-process aggregate. `async_events`, `thread_spans`, `net_spans`, `otel_spans`, and `images`
//! don't carry the column. They keep the `process_id`/`EXISTS` machinery below, which still
//! resolves through `__processes__partitions`, but only for a scan that is *not* a guarded
//! `view_instance(...)` -- see the skip rule in the Branch table below.
//!
//! ## One audience per row, not derived from the process (surviving branches only)
//!
//! `__processes__partitions` (the raw, un-merged `SqlBatchView` partitions `self.processes_source`
//! scans) can carry more than one historical row per `process_id` -- the partition-level
//! `GROUP BY` in `processes_view.rs`'s transform query only collapses rows *within* a partition,
//! not across a long-lived process's entire history. Filtering the views carrying an `audience`
//! column, DDL-defined ones included, one row at a time (the column-filter branch) is sound
//! without any aggregate: every row now carries
//! its own stamp from the moment it was written, so there is nothing to reconcile across rows of
//! the same process. The one exception is a legacy, pre-v8 row with no `audience` column at all --
//! `COALESCE` resolves that to the deployment default at read time. The `net_spans`/
//! `otel_spans`/`images` semi-join branch and the `async_events`/`thread_spans` `EXISTS` branches
//! still resolve through `per_process_audience` -- `Aggregate(GROUP BY process_id, MAX(audience)
//! AS resolved_audience)` -- for a scan of those five that is *not* a guarded instance, since they
//! have no `audience` column of their own to filter directly. A guarded instance skips the
//! predicate entirely (see the Branch table below).
//!
//! ## Branch table
//!
//! Keyed on `MaterializedView::get_view().get_file_schema()`: whether it has an `audience` field,
//! failing that whether it has a `process_id` field -- not on a hardcoded view-set list, so a
//! future view set that gains either column falls into the matching branch automatically:
//!
//! | View set | Branch |
//! |---|---|
//! | `processes`, `streams`, `blocks`, `log_entries`, `measures`, `log_stats`, and any DDL-defined view carrying an `audience` column | `column filter`: direct `audience IN (...)` filter on the view's own column -- no join |
//! | `net_spans`, `otel_spans`, `images`, `async_events`, `thread_spans`, scanned as a guarded, non-`'global'` `view_instance(...)` | `skip`: no predicate at all -- `MaterializedView::instance_is_audience_guarded()` reports `AudienceGuard::authorize_view_instance` already denies an unauthorized caller before any row is read, so the subquery predicates below would be redundant |
//! | `net_spans`, `otel_spans`, `images`, not a guarded instance | `semi-join`: `process_id IN (subquery)` against `per_process_audience` (outer `process_id` cast to `Utf8`: it is `Dictionary(Int32, Utf8)` in these, and nothing coerces an uncorrelated `IN` subquery's join keys once `DecorrelatePredicateSubquery` turns it into a `LeftSemi` join -- the analyzer's own `TypeCoercion` has already run by the time this rule executes) |
//! | `async_events`, not a guarded instance | no `process_id` column either -- `async_events EXISTS`: literal-valued `EXISTS`, keyed on `get_view_instance_id()` (the process_id string, canonicalized -- see `canonical_view_instance_id`) |
//! | `thread_spans`, not a guarded instance | no `process_id` **or** `stream_id` column -- `thread_spans EXISTS`: two-hop literal `EXISTS`, through `streams` (`get_view_instance_id()` is the stream_id string, canonicalized the same way) |
//! | anything in `public_view_sets` | `public`: no predicate at all -- checked before any of the above |
//! | anything else | `analyze()` returns `Err` (`DataFusionError::Plan`) rather than silently leaving the scan unfiltered -- a future view set must add itself to this table, not fall through |
//!
//! Adding a new view set to a `ViewFactory` means adding it to this table too, and to the match
//! arms in `OwnershipRewrite::predicate_for` below.
//!
//! ## No I/O in `analyze()`
//!
//! `in_subquery`/`exists` build uncorrelated subquery `Expr`s; `DecorrelatePredicateSubquery` (an
//! optimizer, not analyzer, rule) turns them into joins later. `processes_source`/`streams_source`
//! come from `MaterializedView::new(..., query_range: None)`, which is synchronous and does no I/O
//! -- the actual `processes`/`streams` scans happen during normal execution, exactly like
//! `TableScanRewrite`'s injected time filter.
//!
//! ## The `audience` column is server-written and authenticated
//!
//! `processes`, `streams`, and `blocks` are each stamped with their own `audience` column at
//! write time, from the authenticated credential's `AuthContext.bound_audience`
//! (`resolve_write_audience`), and a client can neither assert nor forge that stamp
//! (`strip_reserved_properties` drops any client-supplied `micromegas.*` property, and there is
//! no property to re-assert in the first place any more -- the stamp is a physical column, not a
//! property). A credential carrying no bound audience resolves to the deployment default at the
//! HTTP write edge and is stamped with it explicitly, the same as any other audience -- only a
//! legacy, pre-v8 row keeps a NULL column, resolved to the deployment's
//! `MICROMEGAS_DEFAULT_AUDIENCE` where the audience is *read* instead (see `crate::audience`'s
//! module doc). Admin replication (`replication.rs`) now hard-fails on a missing `audience`
//! column rather than ever writing a row with none. Registration
//! (`insert_process`/`register_otel_process`) rejects a same-`process_id`, different-audience
//! re-registration outright -- needed because the OTLP `process_id` derivation formula is public,
//! so without this guard a credential could pre-register (via the native path) the exact
//! `process_id` a victim audience's OTLP producer would later derive, silently exposing that
//! audience's data to the squatter; this is a confidentiality gap, not merely an integrity one.
//! The stream-side mirror, `check_stream_audience_conflict`, closes the equivalent
//! re-pointed-credential gap for `streams`.
//!
//! **What remains open, tracked separately:**
//! - **The five instance-only view sets** (`net_spans`, `otel_spans`, `images`, `async_events`,
//!   `thread_spans`) **and the per-process JIT `view_instance` path** still resolve their audience
//!   *label* through the owning process's/stream's row rather than a genuine per-row column of
//!   their own -- via the call-level guard's instance check (`AudienceGuard::authorize_view_instance`,
//!   `IdKind::ProcessOrStream`) for the guarded `view_instance(...)` scan every production caller
//!   takes, not by a predicate this rule injects (see the skip in the Branch table above). The
//!   cross-audience *injection* scenario (an attacker's block naming a victim's
//!   `process_id`/`stream_id`) is closed for both, as a side effect of the mismatch predicate on
//!   `blocks_view`'s own materialization -- except against a victim whose `processes`/`streams`
//!   row is a legacy, pre-v8 NULL-audience row (next bullet).
//! - **The NULL-anchor window.** A `processes`/`streams` row registered before its ingestion
//!   binary supported per-row stamping keeps `audience = NULL` for its entire remaining life
//!   (rows are immutable, no backfill). The mismatch predicate's NULL-tolerant pass-through lets
//!   a mismatched block through unchecked against such a row for as long as it exists. This is
//!   an accepted, bounded limitation over already-public legacy data, bounded by the deploy-order
//!   requirement (every ingestion replica upgraded) and shrinking as legacy rows age out under
//!   retention.
//!
//! There is no in-product enforcement knob left for either surface; the mitigation is
//! operational -- provision only audience-bound DB-backed ingestion credentials, and don't run
//! ingestion with OIDC or `--disable-auth` alongside them -- see this stage's `CHANGELOG.md`
//! entry.

use super::{materialized_view::MaterializedView, read_scope::ReadScope};
use datafusion::{
    arrow::datatypes::{DataType, Field},
    common::{Column, TableReference, tree_node::Transformed},
    config::ConfigOptions,
    datasource::DefaultTableSource,
    error::DataFusionError,
    functions_aggregate::min_max::max,
    logical_expr::{Filter, LogicalPlan, LogicalPlanBuilder, TableSource},
    optimizer::AnalyzerRule,
    prelude::*,
    scalar::ScalarValue,
};
use std::sync::Arc;
use uuid::Uuid;

/// Injects an audience predicate into every `MaterializedView` scan (see the module doc comment).
pub struct OwnershipRewrite {
    read_scope: ReadScope,
    public_view_sets: Vec<String>,
    processes_source: Arc<dyn TableSource>,
    streams_source: Arc<dyn TableSource>,
}

impl std::fmt::Debug for OwnershipRewrite {
    /// `dyn TableSource` carries no `Debug` impl, so `processes_source`/`streams_source` are
    /// named but not expanded.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("OwnershipRewrite")
            .field("read_scope", &self.read_scope)
            .field("public_view_sets", &self.public_view_sets)
            .field("processes_source", &"<dyn TableSource>")
            .field("streams_source", &"<dyn TableSource>")
            .finish()
    }
}

impl OwnershipRewrite {
    /// `processes_source`/`streams_source` must be built from `MaterializedView::new(...,
    /// query_range: None)` over the raw partitions (equivalent to `__processes__partitions` /
    /// `__streams__partitions`) -- see `query.rs::make_session_context`: the audience lookup must
    /// be time-unbounded, and must not go through the `SqlBatchView`-merged `processes`/`streams`
    /// named tables. Used only by the semi-join/`async_events`/`thread_spans` branches -- the
    /// views carrying an `audience` column, DDL-defined ones included, need neither.
    pub fn new(
        read_scope: ReadScope,
        public_view_sets: Vec<String>,
        processes_source: Arc<dyn TableSource>,
        streams_source: Arc<dyn TableSource>,
    ) -> Self {
        Self {
            read_scope,
            public_view_sets,
            processes_source,
            streams_source,
        }
    }

    /// `Aggregate(GROUP BY process_id, MAX(audience) AS resolved_audience)` over the raw,
    /// time-unbounded `processes` partitions -- built once per `analyze()` call and reused at
    /// every scan site the traversal visits (see the module doc comment). Only reached by the
    /// semi-join/`async_events`/`thread_spans` branches now that `processes` itself carries a
    /// physical `audience` column and filters it directly (the column-filter branch); the
    /// aggregate still exists for those three because they have no `audience` column of their
    /// own, not to reconcile disagreeing rows -- a process's own `audience` column never
    /// disagrees across rows, so `MAX` is just a way to pull the single value out of the group.
    fn per_process_audience(&self) -> datafusion::error::Result<LogicalPlan> {
        LogicalPlanBuilder::scan(
            "__processes__partitions",
            self.processes_source.clone(),
            None,
        )?
        .aggregate(
            vec![col("process_id")],
            vec![max(col("audience")).alias("resolved_audience")],
        )?
        .build()
    }

    /// The caller's audience set. Only ever called after `analyze()` has short-circuited on
    /// `ReadScope::All`, so the `All` arm here is unreachable in practice; it degrades to "no
    /// audiences" rather than panicking if that invariant is ever violated.
    fn audiences(&self) -> &[String] {
        match &self.read_scope {
            ReadScope::All => &[],
            ReadScope::Audiences(a) => a,
        }
    }

    /// `resolved_audience IN (audiences)`, or `lit(false)` when `audiences` is empty -- the
    /// fail-closed reading of "caller has no audiences" rather than emitting `IN ()` and leaving
    /// its behavior to DataFusion (`ReadScope::Audiences` can legitimately resolve to an empty
    /// set -- `public` has no built-in read grant, so a caller matching no grant at all,
    /// including the seeded `('public', 'read', '*')` row, resolves to the empty set, and that
    /// empty set is now reachable in production, not just in tests). No `coalesce` here: the column is
    /// non-null by construction, because the extraction sites already coalesced a
    /// never-stamped process's missing property to the deployment default before it was
    /// materialized.
    fn resolved_predicate(&self) -> Expr {
        let audiences = self.audiences();
        if audiences.is_empty() {
            return lit(false);
        }
        col("resolved_audience").in_list(
            audiences
                .iter()
                .map(|a| lit(ScalarValue::Utf8(Some(a.clone()))))
                .collect(),
            false,
        )
    }

    /// The column-filter branch: `audience IN (caller audiences)`; `false` for an empty set
    /// (fail-closed, as [`Self::resolved_predicate`] already does). The column is declared `NOT NULL` on
    /// `blocks`/`log_entries`/`measures` and non-null by construction (though declared nullable)
    /// on `processes`/`streams`/`log_stats`, so there is no unstamped case either way -- a
    /// DDL-defined view's `audience` column is a documented author obligation to keep non-`NULL`
    /// the same way (see `view_definition.rs`).
    fn audience_column_predicate(&self, table_name: &TableReference, field: &Field) -> Expr {
        let audiences = self.audiences();
        if audiences.is_empty() {
            return lit(false);
        }
        let raw = Expr::Column(Column::new(Some(table_name.clone()), "audience"));
        // This rule runs after DataFusion's own TypeCoercion pass, so the Dictionary(Int32, Utf8)
        // column every audience-carrying view carries must be cast to compare against Utf8
        // literals. The cast is not a pruning barrier: `PruningPredicate` rewrites `cast(col) op
        // lit` by applying the same cast to the column's min/max statistics
        // (`datafusion-pruning`'s `rewrite_expr_to_prunable`), and a Dictionary -> Utf8 cast is on
        // its supported list (`verify_support_type_for_prune` unwraps dictionaries), with `IN`
        // lists up to 20 items expanded to such comparisons. The bare-column arm below is
        // therefore just a no-op-cast skip, kept for a future view that carries plain `Utf8`.
        let lhs = if field.data_type() == &DataType::Utf8 {
            raw
        } else {
            cast(raw, DataType::Utf8)
        };
        lhs.in_list(
            audiences
                .iter()
                .map(|a| lit(ScalarValue::Utf8(Some(a.clone()))))
                .collect(),
            false,
        )
    }

    /// The semi-join branch's shared `IN`-subquery plan: `per_process_audience` filtered by
    /// `resolved_predicate`, projected down to `process_id` alone -- the one column an
    /// `InSubquery` may project.
    fn in_subquery_plan(
        per_process_audience: &LogicalPlan,
        resolved_predicate: &Expr,
    ) -> datafusion::error::Result<Arc<LogicalPlan>> {
        Ok(Arc::new(
            LogicalPlanBuilder::from(per_process_audience.clone())
                .filter(resolved_predicate.clone())?
                .project(vec![col("process_id")])?
                .build()?,
        ))
    }

    /// The `async_events` branch's literal-valued `EXISTS`: `per_process_audience` filtered down
    /// to the single process named by `process_id_literal`, conjuncted with `resolved_predicate`.
    /// No projection: unlike
    /// `InSubquery`, `EXISTS` has no single-column-projection requirement, so the two-column
    /// (`process_id`, `resolved_audience`) shape is fine.
    fn exists_for_process(
        per_process_audience: &LogicalPlan,
        resolved_predicate: &Expr,
        process_id_literal: &str,
    ) -> datafusion::error::Result<Expr> {
        let subquery = LogicalPlanBuilder::from(per_process_audience.clone())
            .filter(
                col("process_id")
                    .eq(lit(process_id_literal))
                    .and(resolved_predicate.clone()),
            )?
            .build()?;
        Ok(exists(Arc::new(subquery)))
    }

    /// The `thread_spans` branch's two-hop literal `EXISTS`: resolve `stream_id_literal` (the
    /// `thread_spans` `view_instance_id`) through `streams` into the process it belongs to, then
    /// apply the same `per_process_audience`/`resolved_predicate` check the column-filter branch
    /// uses directly.
    fn exists_for_stream(
        &self,
        per_process_audience: &LogicalPlan,
        resolved_predicate: &Expr,
        stream_id_literal: &str,
    ) -> datafusion::error::Result<Expr> {
        let subquery =
            LogicalPlanBuilder::scan("__streams__partitions", self.streams_source.clone(), None)?
                .filter(col("stream_id").eq(lit(stream_id_literal)))?
                .join(
                    LogicalPlanBuilder::from(per_process_audience.clone())
                        .filter(resolved_predicate.clone())?
                        .build()?,
                    JoinType::Inner,
                    (vec!["process_id"], vec!["process_id"]),
                    None,
                )?
                .build()?;
        Ok(exists(Arc::new(subquery)))
    }

    /// Parses `view.get_view_instance_id()` as a `Uuid` and returns its canonical
    /// lowercase-hyphenated string form -- the form actually stored in `process_id`/`stream_id`
    /// columns. `Uuid::parse_str` accepts several equivalent textual spellings (uppercase,
    /// hyphen-less, braced) of the same UUID, but
    /// only this canonical form round-trips through the data; using the raw, caller-supplied
    /// spelling directly in a literal comparison would silently fail to match for any other
    /// spelling of a legitimately materialized instance id. Defensive `Err` on parse failure --
    /// the view constructors (`AsyncEventsView::new`/`ThreadSpansView::new`) already validate this
    /// as a UUID at construction time, so this should not happen in practice.
    fn canonical_view_instance_id(
        view_set_name: &Arc<String>,
        view: &Arc<dyn super::view::View>,
    ) -> datafusion::error::Result<String> {
        let raw = view.get_view_instance_id();
        Uuid::parse_str(raw.as_str())
            .map(|uuid| uuid.hyphenated().to_string())
            .map_err(|e| {
                DataFusionError::Plan(format!(
                    "OwnershipRewrite: view set '{view_set_name}' view_instance_id '{raw}' is \
                     not a valid UUID: {e}"
                ))
            })
    }

    /// Builds the predicate to wrap `mat_view`'s scan in. `Ok(None)` means "no predicate at all",
    /// for either of two reasons: the public-view-set skip, or a guarded `view_instance(...)`
    /// scan that `MaterializedView::instance_is_audience_guarded()` reports is already fully
    /// authorized by `AudienceGuard::authorize_view_instance` (the call-level guard); `Err` is the fallback
    /// for a view set matching none of the branches -- see the module doc comment's branch table.
    fn predicate_for(
        &self,
        table_name: &TableReference,
        mat_view: &MaterializedView,
        per_process_audience: &LogicalPlan,
        resolved_predicate: &Expr,
        in_subquery_plan: &Arc<LogicalPlan>,
    ) -> datafusion::error::Result<Option<Expr>> {
        let view = mat_view.get_view();
        let view_set_name = view.get_view_set_name();
        if self
            .public_view_sets
            .iter()
            .any(|s| s.as_str() == view_set_name.as_str())
        {
            return Ok(None);
        }
        // A guarded, non-`'global'` `view_instance(...)` scan is already fully authorized by
        // `AudienceGuard::authorize_view_instance` before any row is read (see
        // `MaterializedView::instance_is_audience_guarded`'s doc comment), so the semi-join/
        // `async_events`/`thread_spans` subquery predicates below are redundant for it. This
        // assumes a guarded instance's rows can never legitimately span more than one process --
        // true for today's five view sets, but a future guarded view set whose instance can
        // legitimately span more than one process would need to revisit this before its arm gets
        // the same skip.
        let skip = mat_view.instance_is_audience_guarded();
        // The column-filter branch: views carrying a physical `audience` column -- processes,
        // streams, blocks, log_entries, measures, log_stats (global and per-process instances
        // alike), and any DDL-defined view carrying the column. Filtered directly, no semi-join,
        // no property_get. Checked ahead of the semi-join branch so a view set that has *both* an
        // `audience` and a `process_id` column (every one of these does) takes this cheaper
        // branch; keyed on schema introspection, so a view set that gains the column later
        // upgrades automatically with no edit here.
        if let Ok(field) = view.get_file_schema().field_with_name("audience") {
            return Ok(Some(self.audience_column_predicate(table_name, field)));
        }
        // Qualified with the outer scan's own table name: once `DecorrelatePredicateSubquery`
        // joins this scan with `in_subquery_plan`'s subquery, the combined schema holds a
        // `process_id` column from both sides, so a bare, unqualified `col("process_id")` is
        // ambiguous and `into_optimized_plan()` fails.
        let outer_process_id = Expr::Column(Column::new(Some(table_name.clone()), "process_id"));
        if view.get_file_schema().field_with_name("process_id").is_ok() {
            // The semi-join branch: process_id-**column** views with no `audience` column of
            // their own -- net_spans, otel_spans, images.
            if skip {
                return Ok(None);
            }
            return Ok(Some(in_subquery(
                cast(outer_process_id, DataType::Utf8),
                in_subquery_plan.clone(),
            )));
        }
        if view_set_name.as_str() == "async_events" {
            // The async_events branch: process-scoped, no process_id column -- the
            // view_instance_id *is* the process_id string (`AsyncEventsView::new` parses it as
            // a `Uuid`). Canonicalized
            // before building the literal -- see `canonical_view_instance_id`'s doc comment for why.
            if skip {
                return Ok(None);
            }
            let process_id_literal = Self::canonical_view_instance_id(&view_set_name, &view)?;
            return Ok(Some(Self::exists_for_process(
                per_process_audience,
                resolved_predicate,
                process_id_literal.as_str(),
            )?));
        }
        if view_set_name.as_str() == "thread_spans" {
            // The thread_spans branch: stream-scoped, no process_id or stream_id column --
            // resolve through `streams`. Same canonicalization as the async_events branch, keyed
            // on `stream_id` instead of `process_id`.
            if skip {
                return Ok(None);
            }
            let stream_id_literal = Self::canonical_view_instance_id(&view_set_name, &view)?;
            return Ok(Some(self.exists_for_stream(
                per_process_audience,
                resolved_predicate,
                stream_id_literal.as_str(),
            )?));
        }
        Err(DataFusionError::Plan(format!(
            "OwnershipRewrite: no audience rule defined for view set '{view_set_name}'"
        )))
    }

    fn rewrite_plan(
        &self,
        plan: LogicalPlan,
        per_process_audience: &LogicalPlan,
        resolved_predicate: &Expr,
        in_subquery_plan: &Arc<LogicalPlan>,
    ) -> datafusion::error::Result<Transformed<LogicalPlan>> {
        let LogicalPlan::TableScan(ts) = &plan else {
            return Ok(Transformed::no(plan));
        };
        let Some(table_source) = ts.source.downcast_ref::<DefaultTableSource>() else {
            return Ok(Transformed::no(plan));
        };
        let Some(mat_view) = table_source
            .table_provider
            .downcast_ref::<MaterializedView>()
        else {
            // Not a MaterializedView (e.g. a table function's output) -- nothing to filter.
            return Ok(Transformed::no(plan));
        };
        let Some(predicate) = self.predicate_for(
            &ts.table_name,
            mat_view,
            per_process_audience,
            resolved_predicate,
            in_subquery_plan,
        )?
        else {
            // No predicate: a public view set, or a guarded `view_instance(...)` scan
            // the call-level guard already authorized.
            return Ok(Transformed::no(plan));
        };
        let filter = Filter::try_new(predicate, Arc::new(plan.clone()))?;
        Ok(Transformed::yes(LogicalPlan::Filter(filter)))
    }
}

impl AnalyzerRule for OwnershipRewrite {
    fn name(&self) -> &str {
        "ownership_rewrite"
    }

    fn analyze(
        &self,
        plan: LogicalPlan,
        _options: &ConfigOptions,
    ) -> datafusion::error::Result<LogicalPlan> {
        if self.read_scope == ReadScope::All {
            // The internal/maintenance marker -- never a ReadPolicy's output. A true no-op, not
            // "filter to a set containing everything" (see the module doc comment).
            return Ok(plan);
        }
        let per_process_audience = self.per_process_audience()?;
        let resolved_predicate = self.resolved_predicate();
        let in_subquery_plan = Self::in_subquery_plan(&per_process_audience, &resolved_predicate)?;
        plan.transform_up_with_subqueries(|node| {
            self.rewrite_plan(
                node,
                &per_process_audience,
                &resolved_predicate,
                &in_subquery_plan,
            )
        })
        .map(|res| res.data)
    }
}