fv_plan/
lib.rs

1//! The DataFusion-backed transform executor.
2//!
3//! Two execution paths:
4//!   • inline row ops (select/rename/drop/filter/applyExpression) — dynamic-typed, row-wise, carrying
5//!     the bespoke value dialect → executed in Rust via the frozen-vectored `fv-value`. Vectored
6//!     against the published contract's `step-kind` conformance vectors.
7//!   • reshapes (join/aggregate/window/union) — cardinality/shape changes that need a relational
8//!     engine → executed on DataFusion (SQL planner over a MemTable), with Value↔Arrow at the edges.
9//!
10//! Governance (gate, lineage, expectations semantics, labels) stays in the control plane; this is
11//! execution only.
12
13/// Every Rust code block in the README is compiled and run as a doctest.
14#[cfg(doctest)]
15#[doc = include_str!("../README.md")]
16pub struct ReadmeDoctests;
17
18pub mod build;
19pub mod compute;
20pub mod convert;
21pub mod decimate_udaf;
22pub mod inline;
23pub mod reshape;
24pub mod row;
25pub mod session;
26pub mod sources;
27pub mod udf;
28
29pub use build::{
30    run_build, run_build_with, run_build_with_native, Build, BuildRecord, ComputeRunner, ExpResult, NoComputeRuntime,
31};
32pub use compute::KineticsRunner;
33pub use inline::{apply_step, apply_steps, Step, StepError};
34pub use reshape::{aggregate, join, union, window, Aggregation, Input, LocalSqlRunner, SqlRunner, WindowFn};
35pub use row::Row;
36
37/// Milliseconds since the Unix epoch (the wall clock a build record is stamped with).
38pub fn now_ms() -> i64 {
39    std::time::SystemTime::now()
40        .duration_since(std::time::UNIX_EPOCH)
41        .map(|d| d.as_millis() as i64)
42        .unwrap_or(0)
43}
44
45/// True if `op` is a whole-transform step (a reshape, raw sql, or a container) — i.e. NOT an inline
46/// row op. Such a step must be a transform's only step. The single source for this classification.
47pub fn is_whole_transform_op(op: &str) -> bool {
48    reshape::NATIVE_OPS.contains(&op) || op == "sql" || op == "container" || op == "wasm"
49}
50
51/// Smoke check that DataFusion is wired (used by the crate's integration test).
52pub async fn datafusion_smoke() -> usize {
53    use datafusion::arrow::array::Int64Array;
54    use datafusion::arrow::datatypes::{DataType, Field, Schema};
55    use datafusion::arrow::record_batch::RecordBatch;
56    use datafusion::prelude::SessionContext;
57    use std::sync::Arc;
58
59    let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, true)]));
60    let batch = RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(vec![1, 2, 3]))]).unwrap();
61    let ctx = SessionContext::new();
62    let df = ctx.read_batch(batch).unwrap();
63    df.collect().await.unwrap().iter().map(|b| b.num_rows()).sum()
64}
fv_plan/lib.rs

fv_plan/
lib.rs
