laminar-db 0.20.1

//! Handle types for query results, source access, and subscriptions.

use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;

use arrow::array::RecordBatch;
use arrow::datatypes::SchemaRef;

use laminar_core::streaming::{Record, Subscription};

use crate::catalog::{ArrowRecord, SourceEntry};
use crate::DbError;

/// Result of executing a SQL statement.
#[derive(Debug)]
pub enum ExecuteResult {
    /// DDL statement completed (CREATE, DROP, ALTER).
    Ddl(DdlInfo),
    /// Query is running, subscribe to results.
    Query(QueryHandle),
    /// Rows were affected (INSERT INTO).
    RowsAffected(u64),
    /// Metadata result (SHOW, DESCRIBE).
    Metadata(RecordBatch),
}

impl ExecuteResult {
    /// Convert to `QueryHandle`, or error if this is not a query result.
    ///
    /// # Errors
    /// Returns `DbError::InvalidOperation` if result is not a query.
    pub fn into_query(self) -> Result<QueryHandle, DbError> {
        match self {
            Self::Query(q) => Ok(q),
            _ => Err(DbError::InvalidOperation(
                "Expected a query result".to_string(),
            )),
        }
    }
}

/// Information about a completed DDL statement.
#[derive(Debug, Clone)]
pub struct DdlInfo {
    /// e.g. `"CREATE SOURCE"`.
    pub statement_type: String,
    /// Object affected.
    pub object_name: String,
}

/// Handle to a running streaming query.
#[derive(Debug)]
pub struct QueryHandle {
    pub(crate) id: u64,
    pub(crate) schema: SchemaRef,
    pub(crate) sql: String,
    pub(crate) subscription: Option<Subscription<ArrowRecord>>,
    pub(crate) active: bool,
}

impl QueryHandle {
    /// Output schema.
    #[must_use]
    pub fn schema(&self) -> &SchemaRef {
        &self.schema
    }

    /// SQL text.
    #[must_use]
    pub fn sql(&self) -> &str {
        &self.sql
    }

    /// Query ID.
    #[must_use]
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Whether the query is still active.
    #[must_use]
    pub fn is_active(&self) -> bool {
        self.active
    }

    pub(crate) fn subscribe_raw(&mut self) -> Result<Subscription<ArrowRecord>, DbError> {
        self.subscription
            .take()
            .ok_or_else(|| DbError::InvalidOperation("Subscription already consumed".to_string()))
    }

    /// Subscribe to typed results. `T` must derive `FromRecordBatch`.
    ///
    /// # Errors
    /// Returns error if subscription was already consumed.
    pub fn subscribe<T: FromBatch>(&mut self) -> Result<TypedSubscription<T>, DbError> {
        let sub = self.subscribe_raw()?;
        Ok(TypedSubscription {
            inner: sub,
            _phantom: PhantomData,
        })
    }

    /// Cancel the query and drop its subscription.
    pub fn cancel(&mut self) {
        self.active = false;
        self.subscription = None;
    }
}

/// Deserialize rows from a `RecordBatch`. Auto-generated by `#[derive(FromRecordBatch)]`.
pub trait FromBatch: Sized {
    /// Single row.
    fn from_batch(batch: &RecordBatch, row: usize) -> Self;
    /// All rows.
    fn from_batch_all(batch: &RecordBatch) -> Vec<Self>;
}

/// Typed subscription that deserializes `RecordBatch` rows.
pub struct TypedSubscription<T: FromBatch> {
    inner: Subscription<ArrowRecord>,
    _phantom: PhantomData<T>,
}

impl<T: FromBatch> TypedSubscription<T> {
    pub(crate) fn from_raw(sub: Subscription<ArrowRecord>) -> Self {
        Self {
            inner: sub,
            _phantom: PhantomData,
        }
    }

    /// Non-blocking poll for the next batch.
    pub fn poll(&mut self) -> Option<Vec<T>> {
        self.inner.poll().map(|batch| T::from_batch_all(&batch))
    }

    /// Blocking receive.
    ///
    /// # Errors
    /// Returns `RecvError` if the channel is closed.
    pub fn recv(&mut self) -> Result<Vec<T>, laminar_core::streaming::RecvError> {
        self.inner.recv().map(|batch| T::from_batch_all(&batch))
    }

    /// Receive with timeout.
    ///
    /// # Errors
    /// Returns `RecvError` on timeout or closed channel.
    pub fn recv_timeout(
        &mut self,
        timeout: Duration,
    ) -> Result<Vec<T>, laminar_core::streaming::RecvError> {
        self.inner
            .recv_timeout(timeout)
            .map(|batch| T::from_batch_all(&batch))
    }

    /// Callback-based drain. Return `false` from `f` to stop.
    pub fn poll_each<F: FnMut(T) -> bool>(&mut self, max_batches: usize, mut f: F) -> usize {
        let mut count = 0;
        for _ in 0..max_batches {
            match self.inner.poll() {
                Some(batch) => {
                    let items = T::from_batch_all(&batch);
                    for item in items {
                        count += 1;
                        if !f(item) {
                            return count;
                        }
                    }
                }
                None => break,
            }
        }
        count
    }

    #[allow(dead_code)]
    pub(crate) fn into_raw(self) -> Subscription<ArrowRecord> {
        self.inner
    }
}

impl<T: FromBatch> std::fmt::Debug for TypedSubscription<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TypedSubscription").finish()
    }
}

/// Typed handle for pushing data into a registered source.
pub struct SourceHandle<T: Record> {
    entry: Arc<SourceEntry>,
    _phantom: PhantomData<T>,
}

impl<T: Record> SourceHandle<T> {
    /// Validates that `T`'s schema matches the source schema.
    pub(crate) fn new(entry: Arc<SourceEntry>) -> Result<Self, DbError> {
        let rust_schema = T::schema();
        let sql_schema = &entry.schema;

        // Validate field count matches
        if rust_schema.fields().len() != sql_schema.fields().len() {
            return Err(DbError::SchemaMismatch(format!(
                "Rust type has {} fields but source '{}' has {} columns",
                rust_schema.fields().len(),
                entry.name,
                sql_schema.fields().len()
            )));
        }

        Ok(Self {
            entry,
            _phantom: PhantomData,
        })
    }

    /// Push a single record.
    ///
    /// # Errors
    /// Returns `StreamingError` if the pipeline is not running.
    #[allow(clippy::needless_pass_by_value)]
    pub fn push(&self, record: T) -> Result<(), laminar_core::streaming::StreamingError> {
        let batch = record.to_record_batch();
        self.entry.push_and_buffer(batch)
    }

    /// Push multiple records, returns count successfully sent.
    pub fn push_batch(&self, records: impl IntoIterator<Item = T>) -> usize {
        const BATCH_SIZE: usize = 1024;
        let mut count = 0;
        let mut buffer = Vec::with_capacity(BATCH_SIZE);

        for record in records {
            buffer.push(record);
            if buffer.len() >= BATCH_SIZE {
                let batch = T::to_record_batch_from_iter(buffer.drain(..));
                if self.push_arrow(batch).is_err() {
                    return count;
                }
                count += BATCH_SIZE;
            }
        }

        if !buffer.is_empty() {
            let len = buffer.len();
            let batch = T::to_record_batch_from_iter(buffer);
            if self.push_arrow(batch).is_ok() {
                count += len;
            }
        }
        count
    }

    /// Push a raw `RecordBatch` (sent to pipeline and buffered for snapshots).
    ///
    /// # Errors
    /// Returns `StreamingError` if the pipeline is not running.
    pub fn push_arrow(
        &self,
        batch: RecordBatch,
    ) -> Result<(), laminar_core::streaming::StreamingError> {
        self.entry.push_and_buffer(batch)
    }

    /// Emit a watermark.
    pub fn watermark(&self, timestamp: i64) {
        self.entry.source.watermark(timestamp);
    }

    /// Current watermark.
    #[must_use]
    pub fn current_watermark(&self) -> i64 {
        self.entry.source.current_watermark()
    }

    /// Buffered record count.
    #[must_use]
    pub fn pending(&self) -> usize {
        self.entry.source.pending()
    }

    /// Buffer capacity.
    #[must_use]
    pub fn capacity(&self) -> usize {
        self.entry.source.capacity()
    }

    /// True when buffer is >80% full.
    #[must_use]
    pub fn is_backpressured(&self) -> bool {
        crate::metrics::is_backpressured(self.pending(), self.capacity())
    }

    /// Source name.
    #[must_use]
    pub fn name(&self) -> &str {
        &self.entry.name
    }

    /// Schema.
    #[must_use]
    pub fn schema(&self) -> &SchemaRef {
        &self.entry.schema
    }

    /// Max out-of-orderness, if configured.
    #[must_use]
    pub fn max_out_of_orderness(&self) -> Option<Duration> {
        self.entry.max_out_of_orderness
    }

    /// Set the event-time column for watermark-based late-row filtering.
    pub fn set_event_time_column(&self, column: &str) {
        self.entry.source.set_event_time_column(column);
    }
}

impl<T: Record> std::fmt::Debug for SourceHandle<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SourceHandle")
            .field("name", &self.entry.name)
            .field("pending", &self.pending())
            .finish()
    }
}

/// Untyped handle for pushing raw `RecordBatch` data.
pub struct UntypedSourceHandle {
    entry: Arc<SourceEntry>,
}

impl UntypedSourceHandle {
    pub(crate) fn new(entry: Arc<SourceEntry>) -> Self {
        Self { entry }
    }

    /// Push a raw `RecordBatch` (sent to pipeline and buffered for snapshots).
    ///
    /// # Errors
    /// Returns `StreamingError` if the pipeline is not running.
    pub fn push_arrow(
        &self,
        batch: RecordBatch,
    ) -> Result<(), laminar_core::streaming::StreamingError> {
        self.entry.push_and_buffer(batch)
    }

    /// Emit a watermark.
    pub fn watermark(&self, timestamp: i64) {
        self.entry.source.watermark(timestamp);
    }

    /// Current watermark.
    #[must_use]
    pub fn current_watermark(&self) -> i64 {
        self.entry.source.current_watermark()
    }

    /// Buffered record count.
    #[must_use]
    pub fn pending(&self) -> usize {
        self.entry.source.pending()
    }

    /// Buffer capacity.
    #[must_use]
    pub fn capacity(&self) -> usize {
        self.entry.source.capacity()
    }

    /// True when buffer is >80% full.
    #[must_use]
    pub fn is_backpressured(&self) -> bool {
        crate::metrics::is_backpressured(self.pending(), self.capacity())
    }

    /// Source name.
    #[must_use]
    pub fn name(&self) -> &str {
        &self.entry.name
    }

    /// Schema.
    #[must_use]
    pub fn schema(&self) -> &SchemaRef {
        &self.entry.schema
    }

    /// Max out-of-orderness, if configured.
    #[must_use]
    pub fn max_out_of_orderness(&self) -> Option<Duration> {
        self.entry.max_out_of_orderness
    }

    /// Set the event-time column for watermark-based late-row filtering.
    pub fn set_event_time_column(&self, column: &str) {
        self.entry.source.set_event_time_column(column);
    }
}

impl std::fmt::Debug for UntypedSourceHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("UntypedSourceHandle")
            .field("name", &self.entry.name)
            .finish()
    }
}

/// Type of a node in the pipeline topology.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PipelineNodeType {
    /// A data source (CREATE SOURCE).
    Source,
    /// A continuous stream (CREATE STREAM).
    Stream,
    /// A data sink (CREATE SINK).
    Sink,
}

/// A node in the pipeline topology graph.
#[derive(Debug, Clone)]
pub struct PipelineNode {
    /// Node name.
    pub name: String,
    /// Node type.
    pub node_type: PipelineNodeType,
    /// Arrow schema (sources only).
    pub schema: Option<SchemaRef>,
    /// SQL definition (streams only).
    pub sql: Option<String>,
}

/// A directed edge in the pipeline topology graph.
#[derive(Debug, Clone)]
pub struct PipelineEdge {
    /// Source node.
    pub from: String,
    /// Target node.
    pub to: String,
}

/// The complete pipeline topology: nodes and edges.
#[derive(Debug, Clone)]
pub struct PipelineTopology {
    /// Nodes.
    pub nodes: Vec<PipelineNode>,
    /// Edges.
    pub edges: Vec<PipelineEdge>,
}

/// Metadata about a registered stream.
#[derive(Debug, Clone)]
pub struct StreamInfo {
    /// Name.
    pub name: String,
    /// Defining SQL query.
    pub sql: Option<String>,
}

/// Information about a registered source.
#[derive(Debug, Clone)]
pub struct SourceInfo {
    /// Name.
    pub name: String,
    /// Schema.
    pub schema: SchemaRef,
    /// Watermark column, if configured.
    pub watermark_column: Option<String>,
}

/// Information about a registered sink.
#[derive(Debug, Clone)]
pub struct SinkInfo {
    /// Name.
    pub name: String,
}

/// Information about a running query.
#[derive(Debug, Clone)]
pub struct QueryInfo {
    /// ID.
    pub id: u64,
    /// SQL text.
    pub sql: String,
    /// Active.
    pub active: bool,
}