knut-thund 0.1.7

//! The **Spark-Connect-SDP lowering** execution backend (feature `spark`).
//!
//! This backend does *not* run a query engine of its own. It lowers the Þund
//! IR to a Spark Declarative Pipelines (SDP) dataflow graph
//! ([`crate::ir::Pipeline::to_sdp`]) and drives a real Spark Connect server
//! through the existing `knut-pipelines` client — i.e. it reuses, rather than
//! reimplements, knut's Spark Connect connector and SDP proto.
//!
//! # Why reuse `knut-pipelines`
//!
//! `knut-pipelines` (feature `grpc`) already vendors the Spark Connect
//! `pipelines.proto` and implements the four-command define-and-run dance over
//! `SparkConnectService.ExecutePlan`:
//!
//! 1. `CreateDataflowGraph` → server `dataflow_graph_id`
//! 2. `DefineOutput` per dataset (table / MV / view / sink)
//! 3. `DefineFlow` per runnable flow (the flow's `Relation`/SQL)
//! 4. `StartRun` → opens the streamed event lifecycle
//!
//! So `SparkBackend` is a thin adaptor: lower the IR
//! ([`crate::ir::Pipeline::to_sdp`]), then — under the `spark` feature — call
//! [`knut_pipelines::PipelinesClient`] to `define_graph` + `start_run`, folding
//! the streamed `PipelineRunEvent`s into [`RunEvent`]s on the [`SparkRun`]. The
//! `define_graph` + `start_run` path is fully wired here; it needs a reachable
//! Spark Connect server to complete (there is no way to fake the cluster), so
//! without one the run returns an honest transport error rather than a stub.
//!
//! # Version requirements (from the SDP research)
//!
//! SDP **GA'd in Apache Spark 4.1.0** (2025-12-16), *not* 4.0. The
//! `pipelines.proto` lives at
//! `sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto` and is
//! absent from branch-4.0. **Spark 4.2+** adds external **sinks**
//! (`create_sink`) and **persistent views**; **expectations** are not yet
//! fully upstreamed into OSS 4.1 — hence [`SparkBackend::capabilities`]
//! reports `expectations: false` by default and the planner refuses to lower
//! a pipeline that relies on them onto Spark (the IR keeps them; the native
//! backend runs them). Always generate the tonic stubs from the proto at the
//! *exact target Spark tag* — `DefineOutput`/`DefineFlow` field names drift
//! between 4.1.0, 4.1.2 and 4.2 master.

use super::{Capabilities, ExecBackend, RunEvent, RunHandle};
use crate::error::{Result, ThundError};
use crate::ir::Pipeline;

/// The Spark-Connect-SDP backend. Holds the Spark Connect URL it lowers onto.
#[derive(Debug, Clone)]
pub struct SparkBackend {
    /// Spark Connect endpoint (`sc://host:15002`).
    pub connect_url: String,
    /// Whether the target server is Spark 4.2+ (enables sinks + persistent
    /// views in the reported capabilities).
    pub spark_4_2_plus: bool,
}

impl SparkBackend {
    /// A backend lowering onto the Spark Connect server at `connect_url`
    /// (assumed Spark 4.1.x by default).
    pub fn new(connect_url: impl Into<String>) -> Self {
        SparkBackend {
            connect_url: connect_url.into(),
            spark_4_2_plus: false,
        }
    }

    /// Declare the target server to be Spark 4.2+, returning `self`.
    pub fn spark_4_2(mut self) -> Self {
        self.spark_4_2_plus = true;
        self
    }

    /// The honest, per-flow record of every streaming/quality knob this Spark
    /// target's SDP lowering (`Pipeline::to_sdp`) will **drop** — the
    /// machine-readable "capability-report the drop" surface (see
    /// [`SdpLoweringReport`]).
    ///
    /// This is the honest half of the Iceberg↔Spark streaming gap. `to_sdp`
    /// lowers the graph *shape* faithfully (datasets/flows/queries), but a Þund
    /// `Flow` carries event-time knobs SDP-over-Connect has no field for —
    /// watermark, window, non-default trigger, non-append output mode — plus
    /// expectations (not upstreamed in OSS 4.1) and, on a 4.1 target, CDC. Those
    /// are silently absent from the lowered `knut_pipelines::Flow`; this method
    /// enumerates exactly which, per flow, so nothing degrades silently. What
    /// this report lists as dropped is precisely what the **native** backend
    /// runs instead.
    ///
    /// A `check`-passing pipeline can still be lossy here: `check` refuses only
    /// what the backend *cannot accept at all* (expectations, CDC on 4.1,
    /// partition transforms); a watermark/window streaming flow *is accepted* by
    /// Spark's SDP — it just runs with Spark's defaults, losing the explicit
    /// event-time semantics. That silent acceptance is the gap this report
    /// makes visible.
    pub fn lowering_report(&self, pipeline: &crate::ir::Pipeline) -> super::SdpLoweringReport {
        use crate::ir::{FlowKind, OutputMode, ScdType, Trigger};

        let mut flow_drops = Vec::new();
        for f in &pipeline.flows {
            let mut dropped = Vec::new();
            match &f.kind {
                FlowKind::Streaming {
                    watermark,
                    trigger,
                    output_mode,
                    window,
                    ..
                } => {
                    if let Some(wm) = watermark {
                        dropped.push(format!(
                            "watermark(event_time=`{}`, allowed_lateness={}ms) — SDP over Connect \
                             has no explicit event-time watermark knob; the stream runs on Spark's \
                             defaults and late rows are not dropped by an allowed-lateness bound. \
                             RUN BY NATIVE: the DataFusion backend evicts late rows past the \
                             allowed-lateness bound (allowed_lateness late-row drop)",
                            wm.event_time_column, wm.allowed_lateness_ms
                        ));
                    }
                    if let Some(w) = window {
                        dropped.push(format!(
                            "window({w:?}) — SDP has no first-class window operator over Connect; \
                             windowing must live inside the flow SQL, so the structured spec is lost. \
                             RUN BY NATIVE: the DataFusion backend closes/evicts event-time windows \
                             (tumbling/hopping/session) on the watermark"
                        ));
                    }
                    match trigger {
                        Trigger::ProcessingTime { interval_ms } => dropped.push(format!(
                            "trigger=ProcessingTime({interval_ms}ms) — SDP streaming tables do not \
                             take an explicit processing-time trigger over Connect. \
                             RUN BY NATIVE: the DataFusion backend fires one trigger per micro-batch \
                             (the wall-clock interval is advisory for an in-process source)"
                        )),
                        // Continuous is the SDP default (no drop); AvailableNow
                        // lowers faithfully as `once = true` (no drop).
                        Trigger::Continuous | Trigger::AvailableNow => {}
                    }
                    if !matches!(output_mode, OutputMode::Append) {
                        dropped.push(format!(
                            "output_mode={output_mode:?} — SDP streaming tables are append-only; \
                             Update/Complete semantics cannot be lowered. \
                             RUN BY NATIVE: the DataFusion backend runs Complete (full table each \
                             trigger) and Update (changed-row upsert delta — not a retraction stream)"
                        ));
                    }
                }
                FlowKind::Cdc { cdc } => {
                    if !self.spark_4_2_plus {
                        dropped.push(
                            "CDC apply-changes — AutoCdcFlowDetails is 4.2-era; this 4.1 target \
                             cannot lower a CDC flow at all (also refused by `check`). \
                             RUN BY NATIVE: the DataFusion backend runs the full apply-changes merge \
                             (INSERT/UPDATE_AFTER upsert, DELETE + apply_as_deletes predicate, \
                             UPDATE_BEFORE pre-image drop)"
                                .to_string(),
                        );
                    } else if matches!(cdc.scd_type, ScdType::Type2) {
                        dropped.push(
                            "SCD Type2 history — OSS SDP AutoCDC carries Type1 (overwrite) only; \
                             the version-history semantics are dropped. \
                             RUN BY NATIVE: the DataFusion backend keeps every version with derived \
                             __start_at/__end_at validity bounds"
                                .to_string(),
                        );
                    }
                }
                FlowKind::Batch => {}
            }
            if !f.expectations.is_empty() {
                dropped.push(format!(
                    "{} expectation(s) — data-quality EXPECT/ON VIOLATION is not upstreamed in OSS \
                     Spark 4.1 SDP (also refused by `check`). \
                     RUN BY NATIVE: the DataFusion backend enforces each expectation \
                     (Warn keeps+records, Drop filters, Fail aborts)",
                    f.expectations.len()
                ));
            }
            if !dropped.is_empty() {
                flow_drops.push(super::SdpFlowDrop {
                    flow: f.name.clone(),
                    dropped,
                });
            }
        }
        super::SdpLoweringReport {
            backend: self.capabilities().name,
            spark_4_2_plus: self.spark_4_2_plus,
            flow_drops,
        }
    }
}

impl ExecBackend for SparkBackend {
    type Run = SparkRun;

    fn capabilities(&self) -> Capabilities {
        Capabilities {
            name: "spark-connect-sdp".into(),
            batch: true,
            streaming: true,
            // SDP has streaming tables but no first-class explicit watermark /
            // window knobs over Connect; those degrade to Spark defaults.
            event_time: false,
            // AutoCDC (`AutoCdcFlowDetails`) is on master / 4.2-era; gate it.
            cdc: self.spark_4_2_plus,
            // Not fully upstreamed in OSS 4.1.
            expectations: false,
            // Parity: incremental watermark/checkpoint state is a native-only
            // engine feature; SDP has no equivalent knob over Connect.
            incremental_state: false,
            // VERIFIED against Spark 4.1.3: SDP's
            // `PartitionHelper#applyPartitioning` maps Seq[Transform] ->
            // Seq[String] and accepts ONLY `IdentityTransform`. Both
            // `PARTITIONED BY (months(event_ts))` and `CLUSTER BY (x)` fail
            // there with PIPELINE_SQL_GRAPH_ELEMENT_REGISTRATION_ERROR
            // "Invalid partitioning transform". Not a 4.1-only gap — 4.2 is the
            // same code, so this is NOT keyed on `spark_4_2_plus`.
            partition_transforms: false,
            // STRUCTURAL false: an SDP flow is a SQL `Relation` and an SDP
            // output is a `TableDetails` (see `knut_pipelines::client`), with
            // no field to carry a Cypher closure / `foreachPartition` / FalkorDB
            // client. A graph load cannot be expressed over Spark-Connect-SDP at
            // all — closing it would need a JVM FalkorDB DataSource V2 + SDP 4.2,
            // a non-Rust subsystem. So `check` refuses graph-sink pipelines
            // loudly (use NativeBackend or the PySpark projection) instead of
            // `DefineOutput`-ing an unresolvable `falkordb-node` format and
            // failing server-side. Not keyed on `spark_4_2_plus`: 4.2 sinks are
            // Spark streaming sinks, still no Cypher shape.
            graph_sinks: false,
            // SDP streaming tables are append-only.
            output_modes: vec!["append".into()],
        }
    }

    fn run(&self, pipeline: &Pipeline) -> Result<Self::Run> {
        self.check(pipeline)?;
        pipeline.validate()?;

        // The faithful part: lower the IR graph shape to SDP. This works on
        // the default build (no `spark`) too — proving the lowering without a
        // live Spark.
        let sdp = pipeline.to_sdp();
        let lowered = sdp
            .validate()
            .map_err(|e| ThundError::Backend(e.to_string()));
        // HONEST STATUS: the IR→SDP lowering + validation genuinely works, so it
        // emits a green row.
        crate::functional_status(
            "knut-thund/backend_spark",
            "lower_to_sdp",
            lowered.is_ok(),
            &self.connect_url,
        );
        lowered?;

        tracing::info!(
            target: "knut_thund::spark",
            pipeline = %pipeline.name,
            connect_url = %self.connect_url,
            datasets = sdp.datasets.len(),
            flows = sdp.flows.len(),
            "spark backend: lowered IR to SDP dataflow graph"
        );

        #[cfg(feature = "spark")]
        {
            exec::run_on_spark(self, pipeline, sdp)
        }
        #[cfg(not(feature = "spark"))]
        {
            let _ = &sdp;
            // Without the `spark` feature the `knut-pipelines` gRPC client is
            // not compiled in. The lowering above is proven; the live run needs
            // `--features spark`. Honest capability gap (not a scaffold TODO).
            crate::functional_status(
                "knut-thund/backend_spark",
                "live_run",
                false,
                "spark feature disabled: Spark Connect client not compiled in",
            );
            Err(ThundError::Backend(
                "spark backend live run requires the `spark` feature (Spark Connect client not \
                 compiled in); IR lowered to SDP graph successfully"
                    .into(),
            ))
        }
    }
}

/// A handle to a Spark-Connect SDP run.
///
/// The `knut-pipelines` client streams the run's `PipelineRunEvent`s to
/// completion; each is folded into a [`RunEvent`] and queued here, drained
/// once by [`RunHandle::poll_events`].
#[derive(Debug, Default)]
pub struct SparkRun {
    /// Events folded from the Spark Connect run stream, in arrival order.
    events: std::collections::VecDeque<RunEvent>,
}

impl RunHandle for SparkRun {
    fn poll_events(&mut self) -> Result<Vec<RunEvent>> {
        Ok(self.events.drain(..).collect())
    }

    fn cancel(&mut self) -> Result<()> {
        Ok(())
    }
}

/// The live Spark Connect driver (feature `spark`).
#[cfg(feature = "spark")]
mod exec {
    use super::{SparkBackend, SparkRun};
    use crate::backend::{RunEvent, RunPhase};
    use crate::error::{Result, ThundError};
    use crate::ir::Pipeline;
    use knut_pipelines::{
        DataflowGraph, PipelineEventSink, PipelineRun, PipelineRunEvent, PipelinesClient,
    };
    use std::collections::VecDeque;

    fn be(ctx: &str, e: impl std::fmt::Display) -> ThundError {
        ThundError::Backend(format!("{ctx}: {e}"))
    }

    /// A sink that folds each `knut-pipelines` run event into a Þund
    /// [`RunEvent`], mapping the SDP severity onto a coarse phase where the
    /// message makes it obvious.
    struct Collector<'a> {
        events: &'a mut VecDeque<RunEvent>,
    }

    impl PipelineEventSink for Collector<'_> {
        fn on_event(&mut self, ev: &PipelineRunEvent) -> knut_pipelines::Result<()> {
            self.events.push_back(RunEvent {
                timestamp: ev.timestamp.clone(),
                element: ev.element.clone(),
                message: ev.message.clone(),
                phase: phase_of(&ev.message),
            });
            Ok(())
        }
    }

    /// Best-effort phase inference from an SDP event message (the proto carries
    /// only timestamp + message; the viewer derives the rest the same way).
    fn phase_of(message: &str) -> Option<RunPhase> {
        let m = message.to_ascii_lowercase();
        if m.contains("fail") || m.contains("error") {
            Some(RunPhase::Failed)
        } else if m.contains("complet") {
            Some(RunPhase::Completed)
        } else if m.contains("running") {
            Some(RunPhase::Running)
        } else if m.contains("queued") {
            Some(RunPhase::Queued)
        } else {
            None
        }
    }

    /// Connect, register the graph, and drive the run to completion, collecting
    /// its events. Synchronous entry point (the [`crate::backend::ExecBackend`]
    /// trait is sync); spins a current-thread Tokio runtime for the gRPC I/O.
    pub(crate) fn run_on_spark(
        backend: &SparkBackend,
        pipeline: &Pipeline,
        sdp: DataflowGraph,
    ) -> Result<SparkRun> {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|e| be("build tokio runtime", e))?;

        let result = rt.block_on(drive(backend, pipeline, &sdp));

        match &result {
            Ok(run) => crate::functional_status(
                "knut-thund/backend_spark",
                "live_run",
                true,
                &format!(
                    "{}: {} event(s) from {}",
                    pipeline.name,
                    run.events.len(),
                    backend.connect_url
                ),
            ),
            Err(e) => crate::functional_status(
                "knut-thund/backend_spark",
                "live_run",
                false,
                &format!("{}: {e}", backend.connect_url),
            ),
        }
        result
    }

    async fn drive(
        backend: &SparkBackend,
        pipeline: &Pipeline,
        sdp: &DataflowGraph,
    ) -> Result<SparkRun> {
        let mut client = PipelinesClient::connect(&backend.connect_url)
            .await
            .map_err(|e| be("connect", e))?;

        let graph_id = client
            .define_graph(sdp)
            .await
            .map_err(|e| be("define_graph", e))?;

        // Spark requires a storage root to actually materialise outputs; with
        // none we do a `dry` validation run rather than provoke a server-side
        // failure — an honest degrade, not a fake success.
        let dry = pipeline.storage.is_none();
        let run_spec = PipelineRun {
            graph_id: Some(graph_id),
            full_refresh_all: true,
            refresh_selection: Vec::new(),
            storage: pipeline.storage.clone(),
            dry,
        };

        let mut events = VecDeque::new();
        if dry {
            events.push_back(RunEvent {
                timestamp: Some(chrono::Utc::now().to_rfc3339()),
                element: None,
                message: "no storage root set — running a dry (validate-only) SDP run".into(),
                phase: Some(RunPhase::Planning),
            });
        }
        {
            let mut sink = Collector {
                events: &mut events,
            };
            client
                .start_run(&run_spec, &mut sink)
                .await
                .map_err(|e| be("start_run", e))?;
        }
        Ok(SparkRun { events })
    }
}