Add EinFold's fused pipeline algorithm to einfold-kernels

crates/einfold-kernels/Cargo.toml

@@ -17,6 +17,9 @@ rust-version.workspace = true
 # without a copy. No query engine.
 arrow = { workspace = true }
 einfold-ir = { workspace = true }
+# Matrix products for the fused pipeline's tiles: ndarray's safe interface
+# to matrixmultiply, which picks AVX2 and FMA kernels at run time.
+ndarray = { version = "0.16", default-features = false, features = ["std"] }
 
 [dev-dependencies]
 # The property test checks the kernels against SQL that DataFusion runs.

crates/einfold-kernels/src/algorithm.rs

@@ -3,16 +3,18 @@
 // SPDX-License-Identifier: Apache-2.0
 
 //! [`fold`] and [`fold_dense`], which run a fold by the best algorithm that
-//! applies, and [`Algorithm`], which says which one ran.
+//! applies, [`Algorithm`], which says which one ran, and [`pipeline`], which
+//! runs a fold pipeline.
 
 use arrow::compute;
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
-use einfold_ir::Fold;
+use einfold_ir::{Fold, Pipeline};
 
+use crate::fused::Piped;
 use crate::options::{DomainRows, Options};
 use crate::relation::Relation;
-use crate::{bad, dense, discriminate, Result};
+use crate::{bad, dense, discriminate, fused, Result};
 
 /// Which algorithm computed a fold.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
@@ -113,3 +115,24 @@ pub fn fold(
         batches,
     })
 }
+
+/// `pipeline` over its producer's operands `q` and `k` and its values `v`, as
+/// batches of `schema` (which [`pipeline_schema`](crate::pipeline_schema)
+/// gives), by the fused algorithm: tile by tile, without the producer's
+/// results ever existing in full. `None` when the algorithm's requirements
+/// don't hold: every input complete, with one row and a non-NULL factor at
+/// every position of its dimensions, keys small non-negative integers, and
+/// at most one condition, between the producer's two outputs. The caller
+/// then runs something else, such as the host's plan.
+///
+/// The result's bits depend only on the inputs' rows, not on how they were
+/// batched or ordered, nor on the number of threads.
+pub fn pipeline(
+    pipeline: &Pipeline,
+    [q, k, v]: [Relation; 3],
+    schema: SchemaRef,
+    options: Options,
+) -> Result<Option<Piped>> {
+    let threads = options.threads.max(1);
+    fused::pipeline(pipeline, [q.batches, k.batches, v.batches], schema, threads)
+}

crates/einfold-kernels/src/dense.rs

@@ -83,7 +83,7 @@ fn worth(cells: usize, rows: usize) -> bool {
 /// plus one), or `None` if it isn't an integer column, or holds a NULL, a
 /// negative key, or one too large for `i64` (an unsigned key of 2⁶³ or more),
 /// or whose extent doesn't fit (a key of `i64::MAX`).
-fn keys(column: &ArrayRef) -> Option<(Int64Array, usize)> {
+pub(crate) fn keys(column: &ArrayRef) -> Option<(Int64Array, usize)> {
     if !column.data_type().is_integer() || column.null_count() > 0 {
         return None;
     }
@@ -104,7 +104,7 @@ fn keys(column: &ArrayRef) -> Option<(Int64Array, usize)> {
 /// column at a time, in 32 bits, which the compiler vectorizes: every
 /// position fits, as the dense algorithm runs only on fewer cells than
 /// `u32::MAX`.
-fn positions(
+pub(crate) fn positions(
     keys: &[(usize, Int64Array)],
     dims: &[usize],
     extent: &[usize],