Add stable multi-key pipeline sorting (#70)

Author: terjekv

crates/hubuum-filter/README.md

@@ -66,7 +66,7 @@ The current native stages provide:
 - typed boolean predicates with explicit casts, null/missing tests, fanout,
   quoted JSON literals, and `NOT`/`AND`/`OR` composition;
 - projection and selector-based value extraction;
-- line or field sorting with string, numeric, and IP casts;
+- stable multi-key sorting with strict casts, fanout reduction, and null order;
 - head, tail, and count;
 - grouping, aggregation, collapse, and array unroll; and
 - JQ-compatible transforms through the in-process `jaq` evaluator.

crates/hubuum-filter/src/eval.rs

@@ -1,11 +1,11 @@
 use regex::Regex;
 
 use crate::error::PipelineError;
-use crate::model::{OutputEnvelope, OutputShape, PipeStage, SortCast};
+use crate::model::{OutputEnvelope, OutputShape, PipeStage};
 use crate::settings::PipelineSettings;
 use crate::verbs::collection::{
     aggregate_envelope, collapse_groups, count_envelope, group_envelope, limit_envelope,
-    sort_envelope, unroll_envelope,
+    sort_columns_envelope, sort_whole_envelope, unroll_envelope,
 };
 use crate::verbs::jq::jq_envelope;
 use crate::verbs::project::{project_envelope, value_envelope};
@@ -53,7 +53,7 @@ impl PipeStage {
             | Self::TypedReject(_)
             | Self::Truthy(_)
             | Self::Columns(_)
-            | Self::SortColumn { .. }
+            | Self::SortColumns(_)
             | Self::Group(_)
             | Self::Aggregate(_)
             | Self::CollapseGroups
@@ -116,15 +116,9 @@ fn apply_semantic_stage(
         PipeStage::Head { count, offset } => limit_envelope(envelope, *count, *offset, false),
         PipeStage::Tail(count) => limit_envelope(envelope, *count, 0, true),
         PipeStage::Count => count_envelope(envelope),
-        PipeStage::SortLines { descending } => {
-            sort_envelope(envelope, None, *descending, SortCast::Auto)
-        }
+        PipeStage::SortLines { descending } => sort_whole_envelope(envelope, *descending),
         PipeStage::Columns(columns) => project_envelope(envelope, columns),
-        PipeStage::SortColumn {
-            selector,
-            descending,
-            cast,
-        } => sort_envelope(envelope, Some(selector), *descending, *cast),
+        PipeStage::SortColumns(spec) => sort_columns_envelope(envelope, spec),
         PipeStage::Group(keys) => group_envelope(envelope, keys),
         PipeStage::Aggregate(spec) => aggregate_envelope(envelope, spec),
         PipeStage::CollapseGroups => collapse_groups(envelope),

crates/hubuum-filter/src/lib.rs

@@ -17,8 +17,8 @@ mod tests;
 pub use error::PipelineError;
 pub use eval::{apply_pipeline, apply_pipeline_with_settings};
 pub use model::{
-    AggregateFunction, AggregateSpec, GroupKey, OutputEnvelope, OutputName, OutputShape, PipeStage,
-    ProjectTerm, SortCast,
+    AggregateFunction, AggregateSpec, GroupKey, NullOrder, OutputEnvelope, OutputName, OutputShape,
+    PipeStage, ProjectTerm, SortCast, SortDirection, SortKey, SortReduction, SortSpec,
 };
 pub use parse::split_pipeline;
 pub use pipeline::Pipeline;

crates/hubuum-filter/src/model.rs

@@ -17,21 +17,12 @@ pub enum PipeStage {
     Truthy(Option<Selector>),
     Reject(String),
     TypedReject(Predicate),
-    Head {
-        count: usize,
-        offset: usize,
-    },
+    Head { count: usize, offset: usize },
     Tail(usize),
     Count,
-    SortLines {
-        descending: bool,
-    },
+    SortLines { descending: bool },
     Columns(Vec<ProjectTerm>),
-    SortColumn {
-        selector: Selector,
-        descending: bool,
-        cast: SortCast,
-    },
+    SortColumns(SortSpec),
     Group(Vec<GroupKey>),
     Aggregate(AggregateSpec),
     CollapseGroups,
@@ -113,7 +104,7 @@ impl PipeStage {
             Self::Head { .. } => "L",
             Self::Tail(_) => "tail",
             Self::Count => "C",
-            Self::SortLines { .. } | Self::SortColumn { .. } => "S",
+            Self::SortLines { .. } | Self::SortColumns(_) => "S",
             Self::Columns(_) => "P",
             Self::Group(_) => "G",
             Self::Aggregate(_) => "A",
@@ -135,7 +126,7 @@ impl PipeStage {
             | Self::Value(_) => STRUCTURED_SHAPES,
             Self::Head { .. } | Self::Tail(_) | Self::SortLines { .. } => COLLECTION_SHAPES,
             Self::Columns(_) => PROJECT_SHAPES,
-            Self::SortColumn { .. } | Self::Unroll(_) => STRUCTURED_COLLECTION_SHAPES,
+            Self::SortColumns(_) | Self::Unroll(_) => STRUCTURED_COLLECTION_SHAPES,
             Self::Group(_) => GROUP_INPUT_SHAPES,
             Self::Aggregate(_) | Self::CollapseGroups => GROUPS_ONLY,
         }
@@ -199,7 +190,7 @@ impl PipeStage {
                     unreachable!("validated input shape")
                 }
             },
-            Self::SortColumn { .. } | Self::Unroll(_) => match input {
+            Self::SortColumns(_) | Self::Unroll(_) => match input {
                 OutputShape::Empty => EMPTY_ONLY,
                 OutputShape::Rows => ROWS_ONLY,
                 OutputShape::Values => VALUES_ONLY,
@@ -389,6 +380,128 @@ pub enum SortCast {
     String,
     Number,
     Ip,
+    Boolean,
+    DateTime,
+    Version,
+    Natural,
+}
+
+impl Display for SortCast {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(match self {
+            Self::Auto => "auto",
+            Self::String => "str",
+            Self::Number => "num",
+            Self::Ip => "ip",
+            Self::Boolean => "bool",
+            Self::DateTime => "datetime",
+            Self::Version => "version",
+            Self::Natural => "natural",
+        })
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum SortDirection {
+    #[default]
+    Ascending,
+    Descending,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum SortReduction {
+    #[default]
+    First,
+    Min,
+    Max,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum NullOrder {
+    First,
+    #[default]
+    Last,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SortKey {
+    selector: Selector,
+    direction: SortDirection,
+    cast: SortCast,
+    reduction: SortReduction,
+    null_order: NullOrder,
+}
+
+impl SortKey {
+    pub fn new(selector: impl Into<String>) -> Result<Self, PipelineError> {
+        Ok(Self {
+            selector: Selector::new(selector)?,
+            direction: SortDirection::Ascending,
+            cast: SortCast::Auto,
+            reduction: SortReduction::First,
+            null_order: NullOrder::Last,
+        })
+    }
+
+    pub fn selector(&self) -> &Selector {
+        &self.selector
+    }
+
+    pub fn direction(&self) -> SortDirection {
+        self.direction
+    }
+
+    pub fn cast(&self) -> SortCast {
+        self.cast
+    }
+
+    pub fn reduction(&self) -> SortReduction {
+        self.reduction
+    }
+
+    pub fn null_order(&self) -> NullOrder {
+        self.null_order
+    }
+
+    pub fn with_direction(mut self, direction: SortDirection) -> Self {
+        self.direction = direction;
+        self
+    }
+
+    pub fn with_cast(mut self, cast: SortCast) -> Self {
+        self.cast = cast;
+        self
+    }
+
+    pub fn with_reduction(mut self, reduction: SortReduction) -> Self {
+        self.reduction = reduction;
+        self
+    }
+
+    pub fn with_null_order(mut self, null_order: NullOrder) -> Self {
+        self.null_order = null_order;
+        self
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SortSpec {
+    keys: Vec<SortKey>,
+}
+
+impl SortSpec {
+    pub fn new(keys: Vec<SortKey>) -> Result<Self, PipelineError> {
+        if keys.is_empty() {
+            return Err(PipelineError::Pipe(
+                "Pipe stage 'S' requires at least one sort key".to_string(),
+            ));
+        }
+        Ok(Self { keys })
+    }
+
+    pub fn keys(&self) -> &[SortKey] {
+        &self.keys
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]