feat: introducing physical plans

Cargo.toml

@@ -26,7 +26,7 @@ aws-config = "0.11.0"
 aws-sdk-s3 = "0.11.0"
 pulsar = { version = "5.1.0", default-features = false, features = ["async-std-runtime", "tokio-runtime", "lz4"] }
 aws-types = "0.11.0"
-bigdecimal = "0.3.0"
+bigdecimal = { vernsion = "0.3.1", features = ["serde"] }
 bincode = "1.3.3"
 bit-set = "0.5.3"
 bitvec = { version = "1.0.1", features = ["serde"] }
@@ -44,6 +44,7 @@ decorum = "0.3.1"
 derive_more = "0.99.17"
 edit-distance = "2.1.0"
 egg = "0.9.3"
+enum-as-inner = "0.6.0"
 enum-map = "2.5.0"
 erased-serde = "0.3.25"
 error-stack = { version = "0.3.1", features = ["anyhow", "spantrace"] }

crates/sparrow-physical/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "sparrow-physical"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+publish = false
+description = """
+Physical execution plans for Kaskada queries.
+"""
+
+[dependencies]
+arrow-schema.workspace = true
+bigdecimal.workspace = true
+enum-as-inner.workspace = true
+serde.workspace = true
+
+[dev-dependencies]
+serde_yaml.workspace = true
+insta.workspace = true
+
+[lib]
+doctest = false

crates/sparrow-physical/src/expr.rs

@@ -0,0 +1,91 @@
+use std::borrow::Cow;
+
+use arrow_schema::DataType;
+
+/// The identifier (index) of an expression.
+#[derive(Debug, serde::Serialize, serde::Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[repr(transparent)]
+#[serde(transparent)]
+pub struct ExprId(usize);
+
+impl From<usize> for ExprId {
+    fn from(value: usize) -> Self {
+        ExprId(value)
+    }
+}
+
+/// A physical expression which describes how a value should be computed.
+///
+/// Generally, each expression computes a column of values from zero or more
+/// input columns. Expressions appear in a variety of places within the steps
+/// that make up a physical plan.
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+
+pub struct Expr {
+    pub kind: ExprKind,
+    /// Arguments to the expression.
+    ///
+    /// These are indices referencing earlier expressions.
+    pub args: Vec<ExprId>,
+    /// The type produced by the expression.
+    pub result_type: DataType,
+}
+
+#[derive(
+    Clone,
+    Debug,
+    Eq,
+    Hash,
+    Ord,
+    PartialEq,
+    PartialOrd,
+    serde::Serialize,
+    serde::Deserialize,
+    enum_as_inner::EnumAsInner,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum ExprKind {
+    /// Apply the named instruction to the given children.
+    Call(Cow<'static, str>),
+    /// Reference an input column by name.
+    Column(String),
+    /// A boolean literal.
+    BooleanLiteral(bool),
+    /// A string literal.
+    StringLiteral(String),
+    /// A numeric literal.
+    ///
+    /// Other primitive literals (such as date times) may be expressed
+    /// using numeric literals with an appropriate datatype.
+    NumericLiteral(bigdecimal::BigDecimal),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_physical_exprs_yaml() {
+        let exprs = vec![
+            Expr {
+                kind: ExprKind::Column("foo".to_owned()),
+                args: vec![],
+                result_type: DataType::Int32,
+            },
+            Expr {
+                kind: ExprKind::Column("bar".to_owned()),
+                args: vec![],
+                result_type: DataType::Int32,
+            },
+            Expr {
+                kind: ExprKind::Call("add".into()),
+                args: vec![0.into(), 1.into()],
+                result_type: DataType::Int32,
+            },
+        ];
+
+        // Use serde_yaml directly, since insta uses a non-standard Yaml formatter.
+        let yaml = serde_yaml::to_string(&exprs).unwrap();
+        insta::assert_snapshot!(yaml)
+    }
+}

crates/sparrow-physical/src/lib.rs

@@ -0,0 +1,19 @@
+#![warn(
+    rust_2018_idioms,
+    nonstandard_style,
+    future_incompatible,
+    clippy::mod_module_files,
+    clippy::print_stdout,
+    clippy::print_stderr,
+    clippy::undocumented_unsafe_blocks
+)]
+
+//! Physical execution plans for Kaskada queries.
+
+mod expr;
+mod plan;
+mod step;
+
+pub use expr::*;
+pub use plan::*;
+pub use step::*;

crates/sparrow-physical/src/plan.rs

@@ -0,0 +1,12 @@
+use crate::Step;
+
+/// A plan is a directed, acyclic graph of steps.
+///
+/// The plan is represented as an array of steps, with each step referencing
+/// it's children (inputs) by index. The array is topologically sorted so that
+/// all children have indices less than the step that references them.
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct Plan {
+    /// The steps in the plan.
+    pub steps: Vec<Step>,
+}

crates/sparrow-physical/src/snapshots/sparrow_physical__expr__tests__physical_exprs_yaml.snap

@@ -0,0 +1,16 @@
+---
+source: crates/sparrow-physical/src/expr.rs
+expression: yaml
+---
+- kind: !column foo
+  args: []
+  result_type: Int32
+- kind: !column bar
+  args: []
+  result_type: Int32
+- kind: !call add
+  args:
+  - 0
+  - 1
+  result_type: Int32
+

crates/sparrow-physical/src/step.rs

@@ -0,0 +1,67 @@
+use arrow_schema::SchemaRef;
+
+/// The identifier (index) of a step.
+
+#[derive(Debug, serde::Serialize, serde::Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[repr(transparent)]
+#[serde(transparent)]
+pub struct StepId(usize);
+
+impl From<usize> for StepId {
+    fn from(value: usize) -> Self {
+        StepId(value)
+    }
+}
+
+/// A single step in the physical plan.
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub struct Step {
+    pub id: StepId,
+    /// The kind of step being performed.
+    pub kind: StepKind,
+    /// Inputs to this step.
+    pub inputs: Vec<StepId>,
+    /// The schema for this step.
+    pub schema: SchemaRef,
+}
+
+/// The kinds of stesp that can occur in the physical plan.
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
+
+pub enum StepKind {
+    /// Scan the given table.
+    Scan {
+        table_name: String,
+    },
+    /// Merge the given relations.
+    Merge,
+    /// Apply a projection to adjust columns in the table.
+    ///
+    /// The output includes the same rows as the input, but with columns
+    /// projected as configured.
+    Project {
+        /// Expressions to apply to compute additional input columns.
+        exprs: Vec<crate::Expr>,
+        /// Indices of expressions to use for the output.
+        ///
+        /// The length should be the same as the number of fields in the schema.
+        outputs: Vec<usize>,
+    },
+    /// Filter the results based on a boolean predicate.
+    Filter {
+        /// Expressions to apply to compute the predicate.
+        ///
+        /// The last expression should be the boolean predicate.
+        exprs: Vec<crate::Expr>,
+    },
+    /// A step that repartitions the output.
+    Repartition {
+        num_partitions: usize,
+        /// Expressions to apply to compute columns which may be referenced by `keys`.
+        exprs: Vec<crate::Expr>,
+        /// Indices of expression columns representing the keys.
+        keys: Vec<usize>,
+    },
+    Error,
+}