uqa-execution 0.5.3

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Revision-safe publication of catalog output metadata.
use super::projection::RegtypeOutputCatalog;
use parking_lot::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
use uqa_sql::SQLError;

#[derive(Default)]
pub struct RegtypeOutputCache {
    entry: Mutex<Option<Arc<RegtypeOutputCatalog>>>,
    initialization: Mutex<()>,
    revision: AtomicU64,
}

impl RegtypeOutputCache {
    /// The session's catalog invalidation generation, including private changes and rollback.
    pub fn revision(&self) -> u64 {
        self.revision.load(Ordering::Acquire)
    }

    pub fn is_populated(&self) -> bool {
        self.entry.lock().is_some()
    }

    pub fn get_or_try_init(
        &self,
        build: impl Fn() -> Result<RegtypeOutputCatalog, SQLError>,
    ) -> Result<Arc<RegtypeOutputCatalog>, SQLError> {
        if let Some(catalog) = self.entry.lock().clone() {
            return Ok(catalog);
        }
        // Refresh may invalidate output metadata while it is being derived; clear never takes this lock.
        let _initialization = self.initialization.lock();
        loop {
            if let Some(catalog) = self.entry.lock().clone() {
                return Ok(catalog);
            }
            let revision = self.revision.load(Ordering::Acquire);
            let built = Arc::new(build()?);
            let mut cache = self.entry.lock();
            if self.revision.load(Ordering::Acquire) != revision {
                drop(cache);
                continue;
            }
            return Ok(cache.get_or_insert(built).clone());
        }
    }
    pub fn clear(&self) {
        let mut entry = self.entry.lock();
        self.revision.fetch_add(1, Ordering::AcqRel);
        entry.take();
    }
}

/// A fallible immutable derivation whose successful value can be borrowed for the catalog's lifetime.
pub(crate) struct CatalogDerivation<T> {
    value: OnceLock<T>,
    initialization: Mutex<()>,
}

impl<T> Default for CatalogDerivation<T> {
    fn default() -> Self {
        Self {
            value: OnceLock::new(),
            initialization: Mutex::new(()),
        }
    }
}

impl<T> CatalogDerivation<T> {
    pub(crate) fn get_or_try_init(
        &self,
        build: impl FnOnce() -> Result<T, SQLError>,
    ) -> Result<&T, SQLError> {
        if let Some(value) = self.value.get() {
            return Ok(value);
        }
        let _initialization = self.initialization.lock();
        if let Some(value) = self.value.get() {
            return Ok(value);
        }
        let value = build()?;
        Ok(self.value.get_or_init(|| value))
    }
}

/// OID selection stops at the first invalid address, while earlier matching entries remain usable.
pub(crate) struct OrderedOidLookup<T> {
    entries: std::collections::BTreeMap<i64, T>,
    error: Option<SQLError>,
}

impl<T> OrderedOidLookup<T> {
    pub(crate) fn build(entries: impl IntoIterator<Item = Result<(i64, T), SQLError>>) -> Self {
        let mut lookup = Self {
            entries: std::collections::BTreeMap::new(),
            error: None,
        };
        for entry in entries {
            match entry {
                Ok((oid, value)) => {
                    lookup.entries.entry(oid).or_insert(value);
                }
                Err(error) => {
                    lookup.error = Some(error);
                    break;
                }
            }
        }
        lookup
    }

    pub(crate) fn get(&self, oid: i64) -> Result<Option<&T>, SQLError> {
        if let Some(value) = self.entries.get(&oid) {
            return Ok(Some(value));
        }
        match &self.error {
            Some(error) => Err(error.clone()),
            None => Ok(None),
        }
    }
}