gnitz-store 0.1.8

//! Opening a relation's stores and entering it in the registry.

use super::dirs::remove_children;
use super::*;
use gnitz_foundation::fault::Seam;

/// `GNITZ_INJECT_TABLE_CREATE_DELAY_MS`: stall a user table's create between its
/// directory and its child subdir, so a concurrent DROP races it.
static TABLE_CREATE_DELAY: Seam = Seam::new("GNITZ_INJECT_TABLE_CREATE_DELAY_MS");

impl RelationRegistry {
    /// Enter a relation, its stores opened fresh.
    pub fn register(&mut self, spec: RelationSpec) -> Result<(), String> {
        let stores = self.build_relation_store(spec, None)?;
        self.enter(spec, stores);
        Ok(())
    }

    /// Enter a view whose rows open from the manifest at `generation` already in
    /// its directory; `Err`, entering nothing, otherwise.
    pub fn reopen_view(&mut self, spec: RelationSpec, generation: u64) -> Result<(), String> {
        let stores = self.build_relation_store(spec, Some(generation))?;
        self.enter(spec, stores);
        Ok(())
    }

    fn enter(&mut self, spec: RelationSpec, (store, feed): (Store, Option<Feed>)) {
        let RelationSpec { id, kind, placement, pk_repeats, .. } = spec;
        let prev = self.tables.insert(
            id,
            Relation {
                id,
                store,
                unsealed: None,
                feed,
                indexes: Vec::new(),
                kind,
                placement,
                pk_repeats,
            },
        );
        debug_assert!(prev.is_none(), "relation {id} registered twice");
    }

    /// This process's stores for a relation: its rows, and a fed view's feed. A
    /// view's rows resume from the manifest at `resume_at`, or are rebuilt.
    pub(super) fn build_relation_store(
        &self,
        spec: RelationSpec,
        resume_at: Option<u64>,
    ) -> Result<(Store, Option<Feed>), String> {
        let RelationSpec { id, kind, schema, placement, .. } = spec;
        let absent = || Ok((Store::Absent(Box::new(schema)), None));
        let (recovery, capacity, feed) = match kind {
            RelationKind::Stream => return absent(),
            // Its store is the relation directory itself, outside the per-slot child
            // layout that a worker-count change relays and reclaims.
            RelationKind::SystemCatalog => {
                let dir = relation_dir(&self.base_dir, id);
                let ram_tier_bytes = self.config.ram_tier_bytes;
                let table = Table::new(&dir, schema, RecoverySource::SalReplay, ram_tier_bytes, None)
                    .map_err(|e| format!("open store '{dir}': {e}"))?;
                return Ok((Store::Held(Box::new(table)), None));
            }
            // The master opens no user store, but it still creates the relation's
            // directory, so the directory exists by the time the DDL is acknowledged.
            _ if !self.residency.owns_stores() => {
                ensure_dir(&relation_dir(&self.base_dir, id))?;
                return absent();
            }
            RelationKind::BaseTable => {
                if let Some(ms) = TABLE_CREATE_DELAY.count() {
                    std::thread::sleep(std::time::Duration::from_millis(ms));
                }
                (RecoverySource::SalReplay, None, None)
            }
            RelationKind::View(p) => {
                if resume_at.is_none() {
                    // The traces a previous compile left here are rebuilt with the rows they
                    // integrate; one the next compile does not declare would never be erased.
                    let slot = self.slot;
                    remove_children(&relation_dir(&self.base_dir, id), |c| {
                        matches!(c.kind, ChildKind::Scratch(_)) && c.slot == slot
                    })
                    .map_err(|e| format!("remove the operator traces of view {id}: {e}"))?;
                }
                (
                    RecoverySource::Rederive { resume_at },
                    p.capacity_bytes(),
                    p.delta_bytes(),
                )
            }
        };
        let rows = self.open_child_as(id, ChildKind::Rows, schema, recovery, capacity)?;
        // Empty at every open: no delta expresses what a boot does to a view, so every cursor restarts.
        let feed = feed.filter(|_| placement.counts_on(self.slot.rank)).map(Feed::new);
        Ok((Store::Held(Box::new(rows)), feed))
    }

    /// [`Self::open_child`], `Err` for a store asked to resume that did not.
    pub(super) fn open_child_as(
        &self,
        id: u64,
        kind: ChildKind<'_>,
        schema: SchemaDescriptor,
        recovery: RecoverySource,
        capacity: Option<u64>,
    ) -> Result<Table, String> {
        let table = self.open_child(id, kind, schema, recovery, capacity)?;
        // A resume that found no manifest at its generation.
        if matches!(recovery, RecoverySource::Rederive { resume_at: Some(_) }) && table.loaded_mark().is_none() {
            return Err(format!(
                "relation {id}: {kind:?} store did not resume from its manifest"
            ));
        }
        Ok(table)
    }

    /// This process's `kind` child store of relation `id`.
    pub(super) fn open_child(
        &self,
        id: u64,
        kind: ChildKind<'_>,
        schema: SchemaDescriptor,
        recovery: RecoverySource,
        capacity: Option<u64>,
    ) -> Result<Table, String> {
        let dir = self.child_dir(id, kind);
        Table::new(&dir, schema, recovery, self.config.ram_tier_bytes, capacity)
            .map_err(|e| format!("open store '{dir}': {e}"))
    }
}