Step the simulation around two known DataFusion bugs it was reporting

.github/workflows/nightly-fuzz.yml

@@ -19,7 +19,7 @@ on:
   workflow_dispatch:
     inputs:
       duration_secs:
-        description: 'Soak budget per region, in seconds'
+        description: 'Soak budget per region, in seconds (at most 3000)'
         required: false
         default: '600'
       base_override:
@@ -43,7 +43,9 @@ jobs:
   soak:
     name: soak region ${{ matrix.region }}
     runs-on: ubuntu-latest
-    timeout-minutes: 40
+    # A build without a cache takes about 15 minutes, and the soak up to 50
+    # more: the job's limit holds both, so its log is uploaded on time.
+    timeout-minutes: 75
     strategy:
       fail-fast: false
       matrix:
@@ -54,6 +56,10 @@ jobs:
         with:
           toolchain: stable
       - uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
+        with:
+          # A soak that finds a failure fails its job, and the build is worth
+          # keeping all the same.
+          cache-on-failure: true
       - name: Build the soak
         run: cargo test -p einfold-datafusion --test simulation --release --no-run
       - name: Run the soak
@@ -65,6 +71,10 @@ jobs:
           REGION: ${{ matrix.region }}
         run: |
           case "$SECS" in ''|*[!0-9]*) echo "duration_secs must be a number"; exit 1;; esac
+          # Checked by length first: the shell's arithmetic wraps on more
+          # digits than a 64-bit number holds.
+          if [ "${#SECS}" -gt 4 ] || [ "$SECS" -gt 3000 ]; then echo "duration_secs must be at most 3000"; exit 1; fi
+          if [ "${#OVERRIDE}" -gt 15 ]; then echo "base_override must have at most 15 digits"; exit 1; fi
           case "$OVERRIDE" in *[!0-9]*) echo "base_override must be a number"; exit 1;; esac
           # Disjoint seed windows that never repeat across runs: runs are
           # 1e8 seeds apart and regions 1e7, and a region explores far fewer.

crates/einfold-datafusion/tests/simulation.rs

@@ -30,7 +30,11 @@
 //!   can answer is one einfold must answer too: moving an expression below a
 //!   join must not make it fail on rows the join drops.
 //! - **`[masks-error]`** einfold fails wherever the host does, rather than
-//!   answering a query whose answer is an error.
+//!   answering a query whose answer is an error, and with the same error.
+//!   Where an error happens can depend on DataFusion's own plan, so where
+//!   only one of them fails, the host on einfold's partitions and batch size
+//!   is the oracle instead, for these and every later check. A panic in
+//!   einfold is always its failure, unless the host panicked the same way.
 //! - **`[schema]`** the same columns, in order: names, types, nullability and
 //!   metadata, both as planned and as the batches carry them.
 //! - **`[rows]`** the same rows, as a multiset. With exact values (small
@@ -103,6 +107,7 @@ use std::io::Write as _;
 use std::sync::Arc;
 
 use datafusion::arrow::array::{Array, ArrayRef, Float64Array, RecordBatch};
+use datafusion::arrow::datatypes::DataType;
 use datafusion::prelude::SessionContext;
 
 use data::{KeyType, Rows, Table, Values};
@@ -139,10 +144,18 @@ enum Known {
     ResetRerunUnderPool,
     /// DataFusion's `EliminateCrossJoin` makes an `=` in the WHERE clause a
     /// key of the join below it, and gives it that join's NULL equality: over
-    /// an `IS NOT DISTINCT FROM` join, NULL then equals NULL. No
-    /// `[partition]` where a mask's `=` is over such a join (einfold#21,
-    /// upstream).
+    /// an `IS NOT DISTINCT FROM` join, NULL then equals NULL. DataFusion
+    /// simplifies `NOT (a <> b)` to `a = b` first, so a mask's `<>` reaches
+    /// it through the mask's negation. No `[partition]` where a mask's `=`
+    /// or `<>` is over such a join (einfold#21, upstream).
     WhereEqualityNullsEqual,
+    /// DataFusion compares dictionary-encoded floats by their bits: of them,
+    /// `0.0` is greater than `-0.0` and the two are separate groups, while
+    /// plain floats' `0.0` and `-0.0` are equal. A mask comparing such a key
+    /// with another float key then splits a group of the unmasked fold
+    /// between the mask and its negation: no `[partition]` where a mask
+    /// compares a dictionary-encoded float (einfold#41, upstream).
+    DictionaryFloatZeros,
 }
 
 const KNOWN: &[Known] = &[
@@ -152,6 +165,7 @@ const KNOWN: &[Known] = &[
     Known::SortMergeFilterProjection,
     Known::ResetRerunUnderPool,
     Known::WhereEqualityNullsEqual,
+    Known::DictionaryFloatZeros,
 ];
 
 impl check::Settings {
@@ -408,6 +422,20 @@ fn register(ctx: &SessionContext, tables: &[Table]) {
     }
 }
 
+/// The host's run on einfold's partitions, batch size and settings.
+async fn control(case: &Case, sql: &str) -> check::Ran {
+    trace("host, on einfold's partitions");
+    let control = check::session(case.threads[1], None, case.batch_sizes[1], case.settings);
+    register(&control, &case.tables);
+    check::run(&control, sql).await
+}
+
+/// An error's message without the values it quotes, such as the text a cast
+/// couldn't read, which depend on which row a plan reached first.
+fn error_kind(e: &str) -> String {
+    e.split('\'').step_by(2).collect::<Vec<_>>().join("'…'")
+}
+
 /// A failure report: the tag, what differed, and how to replay it.
 fn report(case: &Case, tag: &str, what: &str, ours: Option<&check::Ran>) -> String {
     let mut s = format!(
@@ -510,6 +538,18 @@ async fn run_case(case: &Case) -> Outcome {
             }
         }
     }
+    // An error can depend on the plan, not only on einfold: a cast of text
+    // that doesn't read as a number fails only where a row reaches it, and
+    // DataFusion's own plan on other partitions can reach rows the host's
+    // plan didn't. So where only one of the host and einfold fails, the host
+    // on einfold's partitions and batch size is the oracle instead: einfold
+    // must succeed where it does, match its answer, and fail where it does
+    // with the same error.
+    let controlled = theirs.result.is_ok() != mine.result.is_ok();
+    let theirs = match controlled {
+        true => control(case, sql).await,
+        false => theirs,
+    };
     let (want, got) = match (&theirs.result, &mine.result) {
         (Ok(w), Ok(g)) => (w, g),
         // Under a small memory pool, either side may run out where the other
@@ -545,6 +585,23 @@ async fn run_case(case: &Case) -> Outcome {
             );
             return out;
         }
+        // A panic in einfold is einfold's, unless the host panicked the
+        // same way; and the control must fail as einfold did. Running out of
+        // memory is tallied, as above, whatever the other side's error.
+        (Err(w), Err(g))
+            if error_kind(w) != error_kind(g)
+                && (controlled || g.starts_with("panicked"))
+                && !out_of_memory(w)
+                && !out_of_memory(g) =>
+        {
+            out.host_failed = true;
+            fail(
+                "error",
+                format!("einfold failed unlike the host: {g}\n  the host: {w}"),
+                Some(&mine),
+            );
+            return out;
+        }
         (Err(_), Err(_)) => {
             out.host_failed = true;
             return out;
@@ -582,10 +639,18 @@ async fn run_case(case: &Case) -> Outcome {
     // case are, between them, exactly the unmasked fold's, with the same
     // values.
     if let (Some(m), Values::Exact) = (&case.query.mask, case.values) {
-        let nulls_equal = m.condition.contains(" = ")
+        let nulls_equal = (m.condition.contains(" = ") || m.condition.contains(" <> "))
             && case.query.sql.contains("IS NOT DISTINCT FROM")
             && avoids(Known::WhereEqualityNullsEqual);
-        if m.partitions && !nulls_equal {
+        // The mask's columns, as `l.a`: the tables' aliases are their names.
+        let dictionary_floats = (case.tables.iter())
+            .flat_map(|t| t.schema.fields().iter().map(move |f| (t.name, f)))
+            .any(|(table, f)| {
+                matches!(f.data_type(), DataType::Dictionary(_, v) if **v == DataType::Float64)
+                    && m.condition.contains(&format!("{table}.{}", f.name()))
+            })
+            && avoids(Known::DictionaryFloatZeros);
+        if m.partitions && !nulls_equal && !dictionary_floats {
             trace("einfold, partition");
             let with = |c: &str| m.template.replace("{MASK}", c);
             let all = check::run(&host, &with("TRUE")).await.result;
@@ -855,6 +920,21 @@ fn simulation() {
     }
 }
 
+/// Two runs of the same panic fail alike, with its message: the host's and
+/// einfold's panics are compared by what their failures say.
+#[test]
+fn panics_fail_alike() {
+    let rt = runtime();
+    let fail = || {
+        rt.block_on(async {
+            let joined = tokio::spawn(async { panic!("boom") }).await;
+            check::panicked(joined.expect_err("a panic"))
+        })
+    };
+    assert_eq!(fail(), "panicked: boom");
+    assert_eq!(fail(), fail());
+}
+
 /// Fails if the generator stopped reaching what the properties are about.
 fn check_coverage(tally: &Tally) {
     let fired = |s: Shape| tally.get(s, 1);

crates/einfold-datafusion/tests/simulation/check.rs

@@ -157,7 +157,7 @@ pub async fn run(ctx: &SessionContext, sql: &str) -> Ran {
     .await
     {
         Ok(r) => r.map_err(|e| e.to_string()),
-        Err(e) => Err(format!("panicked: {e}")),
+        Err(e) => Err(panicked(e)),
     };
     Ran {
         result,
@@ -166,6 +166,20 @@ pub async fn run(ctx: &SessionContext, sql: &str) -> Ran {
     }
 }
 
+/// A panicked task's failure, as its panic's message: Tokio's own text for
+/// it names the task's ID, which differs between runs of the same panic.
+pub fn panicked(e: tokio::task::JoinError) -> String {
+    match e.try_into_panic() {
+        Ok(payload) => {
+            let message = (payload.downcast_ref::<&str>().map(|m| m.to_string()))
+                .or_else(|| payload.downcast_ref::<String>().cloned())
+                .unwrap_or_else(|| "a panic without a message".to_string());
+            format!("panicked: {message}")
+        }
+        Err(e) => format!("panicked: {e}"),
+    }
+}
+
 /// Runs a plan that already ran once, again, after resetting its state as
 /// DataFusion asks of a plan it runs twice (`reset_plan_states`, which a
 /// recursive query uses for each iteration).
@@ -181,7 +195,7 @@ pub async fn rerun(ctx: &SessionContext, ran: &Ran) -> Option<Result<Vec<RecordB
         tokio::spawn(async move { datafusion::physical_plan::collect(plan, task).await });
     Some(match collected.await {
         Ok(r) => r.map_err(|e| e.to_string()),
-        Err(e) => Err(format!("panicked: {e}")),
+        Err(e) => Err(panicked(e)),
     })
 }