datum-core 0.10.4

//! Akka-style stream/stage planner attributes.
//!
//! [`Attributes`] is a small, immutable, cloneable bag of [`Attribute`]s attached to a stream
//! blueprint or a graph stage. Attributes carry validated planning intent and introspection:
//! names, input-buffer and dispatcher hints, batching/concurrency hints, fusion constraints,
//! metrics level, and kernel backend preference. A1 activates the existing execution machinery for
//! [`Fusion::TypedOnly`] and [`Fusion::ErasedOnly`]; [`Attribute::BatchSize`],
//! [`Attribute::Parallelism`], [`Attribute::MetricsLevel`], and [`Attribute::KernelBackend`] are
//! validated, report-visible plan inputs whose execution semantics are consumed by later M14
//! metrics/kernel rungs.
//!
//! Combine two bags with [`Attributes::and`]: the **argument** takes precedence over the receiver,
//! so the most-recently-applied (most specific) value wins the accessor lookups
//! ([`Attributes::name`], [`Attributes::input_buffer_hint`], [`Attributes::dispatcher_hint`]).
//!
//! Mirrors `akka.stream.Attributes`; `Attribute`/`Attributes` are re-exported from the crate root.

use std::sync::Arc;

use thiserror::Error;

/// Validation error returned by fallible [`Attributes`] constructors.
#[derive(Clone, Debug, PartialEq, Eq, Error)]
pub enum AttributeError {
    #[error("input buffer initial and max must be nonzero")]
    ZeroInputBuffer,
    #[error("input buffer initial ({initial}) must be <= max ({max})")]
    InputBufferMinExceedsMax { initial: usize, max: usize },
    #[error("batch size must be greater than zero")]
    ZeroBatchSize,
    #[error("parallelism must be greater than zero")]
    ZeroParallelism,
}

/// Fallible result for validated attribute constructors.
pub type AttributeResult<T> = Result<T, AttributeError>;

/// Fusion preference for a graph node.
///
/// `TypedOnly` and `ErasedOnly` are active in M14-A1. `Auto` leaves the executor free to choose the
/// best available tier. Datum deliberately does not insert async boundaries from this attribute in
/// A1; explicit [`crate::AsyncBoundary`] stages remain the boundary mechanism.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
pub enum Fusion {
    /// Let the planner select the fastest supported tier.
    #[default]
    Auto,
    /// Require a typed executor tier and fail materialization if the graph would fall to erased.
    TypedOnly,
    /// Force the erased executor tier and report the tier change.
    ErasedOnly,
}

/// Requested metrics detail for a graph node.
///
/// Metrics collection is implemented by the M14 metrics rung. In A1 this is a validated,
/// report-visible planner input only.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
pub enum MetricsLevel {
    /// No per-node metrics.
    #[default]
    Off,
    /// Element/batch/row counters.
    Counts,
    /// Counters plus processing-time measurements.
    Timing,
    /// Counters, timing, and backpressure/stall observations.
    Stalls,
}

/// Kernel backend preference for lowerable batch/scalar nodes.
///
/// Backend selection is implemented by the M14 kernel rungs. In A1 this is a validated,
/// report-visible planner input only.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
pub enum KernelBackend {
    /// Let the planner choose the backend for a lowerable node.
    Auto,
    /// Use Datum's Rust/Arrow backend.
    #[default]
    ArrowRust,
    /// Use a Python/C++ Arrow backend when that backend is available and selected.
    Python,
}

/// A single validated stream/stage planner attribute.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Attribute {
    /// Human-readable stage/stream name (used in diagnostics).
    Name(Arc<str>),
    /// Requested input-buffer sizing hint (`initial` ≤ `max`).
    InputBuffer { initial: usize, max: usize },
    /// Dispatcher name hint for where the stage's work should run.
    Dispatcher(Arc<str>),
    /// Preferred batch size for batch-producing or batch-coalescing stages.
    BatchSize(usize),
    /// Per-node concurrency limit where semantics allow parallel work.
    Parallelism(usize),
    /// Fusion constraint/preference.
    Fusion(Fusion),
    /// Requested metrics detail.
    MetricsLevel(MetricsLevel),
    /// Kernel backend preference.
    KernelBackend(KernelBackend),
}

/// An ordered, immutable collection of [`Attribute`]s.
///
/// Lookups return the first matching attribute; [`Attributes::and`] prepends the argument's
/// attributes so the more specific bag wins. Construction never starts any work.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Attributes {
    attribute_list: Arc<[Attribute]>,
}

impl Attributes {
    /// An empty attribute set (the `Default`).
    #[must_use]
    pub fn none() -> Self {
        Self::default()
    }

    /// A set holding a single [`Attribute::Name`].
    #[must_use]
    pub fn named(name: impl Into<String>) -> Self {
        Self {
            attribute_list: Arc::from([Attribute::Name(Arc::from(name.into()))]),
        }
    }

    /// A set holding a single [`Attribute::InputBuffer`] hint.
    #[must_use]
    pub fn input_buffer(initial: usize, max: usize) -> Self {
        Self::try_input_buffer(initial, max).expect("invalid input buffer attribute")
    }

    /// Fallible constructor for a single [`Attribute::InputBuffer`] hint.
    pub fn try_input_buffer(initial: usize, max: usize) -> AttributeResult<Self> {
        validate_input_buffer(initial, max)?;
        Ok(Self {
            attribute_list: Arc::from([Attribute::InputBuffer { initial, max }]),
        })
    }

    /// A set holding a single [`Attribute::Dispatcher`] hint.
    #[must_use]
    pub fn dispatcher(name: impl Into<String>) -> Self {
        Self {
            attribute_list: Arc::from([Attribute::Dispatcher(Arc::from(name.into()))]),
        }
    }

    /// A set holding a single [`Attribute::BatchSize`] hint.
    ///
    /// A1 records this in execution reports; batch execution semantics are added by later M14
    /// rungs.
    #[must_use]
    pub fn batch_size(size: usize) -> Self {
        Self::try_batch_size(size).expect("invalid batch size attribute")
    }

    /// Fallible constructor for a single [`Attribute::BatchSize`] hint.
    pub fn try_batch_size(size: usize) -> AttributeResult<Self> {
        if size == 0 {
            return Err(AttributeError::ZeroBatchSize);
        }
        Ok(Self {
            attribute_list: Arc::from([Attribute::BatchSize(size)]),
        })
    }

    /// A set holding a single [`Attribute::Parallelism`] hint.
    ///
    /// A1 records this in execution reports; per-node parallel execution semantics are added by
    /// later M14 rungs.
    #[must_use]
    pub fn parallelism(parallelism: usize) -> Self {
        Self::try_parallelism(parallelism).expect("invalid parallelism attribute")
    }

    /// Fallible constructor for a single [`Attribute::Parallelism`] hint.
    pub fn try_parallelism(parallelism: usize) -> AttributeResult<Self> {
        if parallelism == 0 {
            return Err(AttributeError::ZeroParallelism);
        }
        Ok(Self {
            attribute_list: Arc::from([Attribute::Parallelism(parallelism)]),
        })
    }

    /// A set holding a single [`Attribute::Fusion`] preference.
    #[must_use]
    pub fn fusion(fusion: Fusion) -> Self {
        Self {
            attribute_list: Arc::from([Attribute::Fusion(fusion)]),
        }
    }

    /// A set holding a single [`Attribute::MetricsLevel`] preference.
    #[must_use]
    pub fn metrics_level(level: MetricsLevel) -> Self {
        Self {
            attribute_list: Arc::from([Attribute::MetricsLevel(level)]),
        }
    }

    /// A set holding a single [`Attribute::KernelBackend`] preference.
    #[must_use]
    pub fn kernel_backend(backend: KernelBackend) -> Self {
        Self {
            attribute_list: Arc::from([Attribute::KernelBackend(backend)]),
        }
    }

    /// Combine two sets; the attributes of `other` take precedence on accessor lookups.
    #[must_use]
    pub fn and(mut self, other: Self) -> Self {
        if other.attribute_list.is_empty() {
            return self;
        }
        if self.attribute_list.is_empty() {
            return other;
        }
        let mut combined =
            Vec::with_capacity(other.attribute_list.len() + self.attribute_list.len());
        combined.extend(other.attribute_list.iter().cloned());
        combined.extend(self.attribute_list.iter().cloned());
        self.attribute_list = Arc::from(combined.into_boxed_slice());
        self
    }

    /// The underlying attributes, in lookup order (most specific first).
    #[must_use]
    pub fn attribute_list(&self) -> &[Attribute] {
        &self.attribute_list
    }

    /// Validate every attribute in this bag.
    pub fn validate(&self) -> AttributeResult<()> {
        for attribute in self.attribute_list.iter() {
            match attribute {
                Attribute::InputBuffer { initial, max } => validate_input_buffer(*initial, *max)?,
                Attribute::BatchSize(0) => return Err(AttributeError::ZeroBatchSize),
                Attribute::Parallelism(0) => return Err(AttributeError::ZeroParallelism),
                Attribute::Name(_)
                | Attribute::Dispatcher(_)
                | Attribute::BatchSize(_)
                | Attribute::Parallelism(_)
                | Attribute::Fusion(_)
                | Attribute::MetricsLevel(_)
                | Attribute::KernelBackend(_) => {}
            }
        }
        Ok(())
    }

    /// The first [`Attribute::Name`], if any.
    #[must_use]
    pub fn name(&self) -> Option<&str> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::Name(name) => Some(name.as_ref()),
                _ => None,
            })
    }

    /// The first [`Attribute::InputBuffer`] hint as `(initial, max)`, if any.
    #[must_use]
    pub fn input_buffer_hint(&self) -> Option<(usize, usize)> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::InputBuffer { initial, max } => Some((*initial, *max)),
                _ => None,
            })
    }

    /// The first [`Attribute::Dispatcher`] hint, if any.
    #[must_use]
    pub fn dispatcher_hint(&self) -> Option<&str> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::Dispatcher(name) => Some(name.as_ref()),
                _ => None,
            })
    }

    /// The first [`Attribute::BatchSize`] hint, if any.
    #[must_use]
    pub fn batch_size_hint(&self) -> Option<usize> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::BatchSize(size) => Some(*size),
                _ => None,
            })
    }

    /// The first [`Attribute::Parallelism`] hint, if any.
    #[must_use]
    pub fn parallelism_hint(&self) -> Option<usize> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::Parallelism(parallelism) => Some(*parallelism),
                _ => None,
            })
    }

    /// The first [`Attribute::Fusion`] preference, if any.
    #[must_use]
    pub fn fusion_hint(&self) -> Option<Fusion> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::Fusion(fusion) => Some(*fusion),
                _ => None,
            })
    }

    /// The first [`Attribute::MetricsLevel`] preference, if any.
    #[must_use]
    pub fn metrics_level_hint(&self) -> Option<MetricsLevel> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::MetricsLevel(level) => Some(*level),
                _ => None,
            })
    }

    /// The first [`Attribute::KernelBackend`] preference, if any.
    #[must_use]
    pub fn kernel_backend_hint(&self) -> Option<KernelBackend> {
        self.attribute_list
            .iter()
            .find_map(|attribute| match attribute {
                Attribute::KernelBackend(backend) => Some(*backend),
                _ => None,
            })
    }

    /// Return a copy with a [`Attribute::Name`] prepended so it wins [`Attributes::name`].
    #[must_use]
    pub fn with_name(mut self, name: impl Into<String>) -> Self {
        let mut attributes = Vec::with_capacity(self.attribute_list.len() + 1);
        attributes.push(Attribute::Name(Arc::from(name.into())));
        attributes.extend(self.attribute_list.iter().cloned());
        self.attribute_list = Arc::from(attributes.into_boxed_slice());
        self
    }
}

fn validate_input_buffer(initial: usize, max: usize) -> AttributeResult<()> {
    if initial == 0 || max == 0 {
        return Err(AttributeError::ZeroInputBuffer);
    }
    if initial > max {
        return Err(AttributeError::InputBufferMinExceedsMax { initial, max });
    }
    Ok(())
}