flowlog: native :merge for term-mode congruence (--native-merge)

Port the non-proof half of egglog PR egraphs-good#933 to the FlowLog backend:
replace the rule-encoded @congruence_rule* with an FD view keyed on
children + a UnionId-style side-effecting merge that unions colliding
outputs at FD-conflict time (drained at the iteration boundary). Behind
a --native-merge flag (FlowLog-only, term mode, implies --native-uf);
off by default with byte-identical rule encoding. Bit-exact to flowlog's
rule encoding; agent-measured ~1.16x (1.28x with --fast-rebuild) faster
than --native-uf on math-microbenchmark. Also fixes a pre-existing
clone_boxed panic on --native-uf --flowlog/feldera.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: oflatt

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased] - ReleaseDate
 
+- **`--native-merge`: inline congruence via a native backend merge (FlowLog, term mode).** Ports the non-proof half of egraphs-good/egglog PR #933 to the FlowLog dataflow backend: instead of desugaring congruence into a seminaive `@congruence_rule*` self-join run inside the per-`(run)` rebuild fixpoint, a CONSTRUCTOR's `@<F>View` is keyed on its CHILDREN ONLY (an FD view `(children) -> eclass`) with a side-effecting `MergeFn::UnionId` merge that, on an FD conflict (same children, two eclasses), UNIONS the two eclasses into the union-find and keeps the min — congruence done INLINE at insert time, and the `@congruence_rule*` rule dropped entirely. Implies `--native-uf` (the injected union edge goes into FlowLog's in-core union-find via `enqueue_union`, drained at the iteration boundary like every other union; the displaced-id signal drives the existing rebuild to re-canonicalize children, which re-triggers the merge for transitive congruence). The frontend records each FD view's name (`EncodingState::native_merge_views`) so lowering substitutes `MergeFn::UnionId` and `register_native_merge_view` tells the backend which UF owns the eclass column; the view body/set/delete/subsume forms switch to the `(= eclass (@view children))` shape via the new `view_body_atom`/`view_delete` helpers. Bit-exact with the bridge `--term-encoding` oracle (math-microbenchmark runs 1/3/5/7, hand-written congruence tests, a Herbie sample file) and faster than the rule-encoded congruence (math run 9: 1.14x vs `--native-uf`, 1.21x vs `--native-uf --fast-rebuild`). FlowLog-only, term mode, eq-sort-output constructors with ≥1 input; 0-input constructors (globals) keep the baseline all-columns view. Off by default — the rule-encoded congruence is byte-identical when off. Also fixes a pre-existing `--native-uf --flowlog`/`--feldera` panic (`clone_boxed` unimplemented) by skipping the cli `with_term_encoding_enabled` clone for those backends (they provision the typechecker at construction). (`src/proofs/proof_encoding.rs`, `src/lib.rs`, `src/cli.rs`, `egglog-bridge-flowlog/src/lib.rs`, `egglog-backend-trait/src/lib.rs`, `egglog-experimental/src/main.rs`.)
+
 - **Proof mode now supports canonicalize-at-creation, and `--fast-rebuild` works with `--proofs` (native bridge).** Previously canon-at-creation and the `--fast-rebuild` `δview ⋈ uf_old` drop were gated off in proof mode, because the proof-mode flat UF index `@UF_Sf` is `(S) -> Pair(S, Proof)` (key/output types differ, so `DefaultVal::Identity` is unavailable and a lookup of a just-minted key fails). Proof-mode canon now goes through two per-eq-sort `ReadPrim`s declared via a new `:internal-uf-pair-prims` annotation on the `@UFPair_S` sort (so they survive the desugar round-trip): `@uf_find_leader_S` (`(S) -> S`, lookup-or-self leader for child columns) and `@uf_find_or_refl_S` (`(S Proof) -> @UFPair_S`, lookup-or-`(pair x fallback)` for the representative column, no row inserted — bit-exact tuple counts). The constructor term is still built from the original (non-canonical) args so the rule proof's claimed equality matches; the view's child/representative columns are canonicalized and the per-child `child = leader` proofs are threaded into the view proof via `Congr`/`Trans(Sym(rep_pf), …)`. Bridge-only (gated on `Backend::supports_inline_table_lookups`); constructors only (custom `:merge` function views keep the baseline rebuild-rule canonicalization and their `δview` term under fast-rebuild). Bit-exact with proof-mode full-rebuild, and proofs still check. (`src/proofs/proof_encoding.rs`, `src/typechecking.rs`.)
 
 - **Fixed: `egglog-experimental` ignored `--threads`/`-j` and silently ran rayon at full width.** `egglog::cli` configures the rayon global pool via `set_num_threads` → `build_global`, but only after it's handed an already-constructed egraph; the experimental wrapper main builds the (experimental/backend) egraph first, and that construction touches rayon — lazily initializing the global pool at its default (all logical CPUs). So `cli`'s later `build_global(1)` failed (`GlobalPoolAlreadyInitialized`) and `--threads 1` was a no-op (e.g. ~257 threads spawned instead of running serial). Fix: the experimental main now calls `set_num_threads` (parsing `--threads`/`-j`, default 1) BEFORE constructing the egraph, and `EGraph::set_num_threads` is now idempotent (a `Once` guard — first call wins, later calls silently ignored per its doc, no spurious warning). The plain `egglog` binary was unaffected (it constructs `EGraph::default()`, which doesn't touch rayon, so its `set_num_threads` already won). (egglog-experimental + `EGraph::set_num_threads`.)

egglog-backend-trait/src/lib.rs

@@ -295,6 +295,16 @@ pub trait Backend: Send + Sync {
     /// rebuild; the dataflow/SQL backends drive their own host-pass rebuild.
     fn register_nativerb_view(&mut self, _uf_func: FunctionId, _view_func: FunctionId) {}
 
+    /// `--native-merge` (term-encoding native-UF, non-proof): associate an
+    /// FD-keyed constructor view (`view_func`, keyed `(children) -> eclass`) with
+    /// the union-find (`uf_func`) that owns its eclass (OUTPUT) column. The view's
+    /// `MergeFn::UnionId` merge then injects a union into this UF on an FD conflict
+    /// (same children, two eclasses) — congruence done INLINE at insert time
+    /// instead of by a self-join rule. The frontend calls this before each run for
+    /// each not-yet-registered native-merge view. Default no-op: only backends
+    /// that inject the union on FD-conflict (currently FlowLog) implement it.
+    fn register_native_merge_view(&mut self, _uf_func: FunctionId, _view_func: FunctionId) {}
+
     /// Register a union-find-backed function (see upstream PR #782).
     ///
     /// The function has schema `(S) S` for an EqSort `S` and records leader

egglog-bridge-flowlog/src/lib.rs

@@ -304,6 +304,13 @@ pub struct EGraph {
     /// non-triangle rules are unaffected. Off by default — when off the FlowLog
     /// join path is byte-identical to the pre-WCOJ behavior.
     pub(crate) wcoj_enabled: bool,
+    /// `--native-merge`: maps an FD-keyed CONSTRUCTOR view's [`FunctionId`] to the
+    /// [`FunctionId`] of the union-find (`@UF_Sf`) that owns its eclass (OUTPUT)
+    /// column. Populated by [`Backend::register_native_merge_view`]. When a view
+    /// in this map has an FD conflict (`resolve_merge` finds two eclasses for one
+    /// children key), the loser is `enqueue_union`-ed with the winner into this UF
+    /// — congruence done INLINE at insert time instead of by a self-join rule.
+    pub(crate) native_merge_uf: HashMap<FunctionId, FunctionId>,
 }
 
 impl Default for EGraph {
@@ -377,6 +384,7 @@ impl EGraph {
             native_uf_disp_rel: HashMap::new(),
             native_uf_rebuild_dd_ir: HashMap::new(),
             wcoj_enabled: false,
+            native_merge_uf: HashMap::new(),
         }
     }
 
@@ -658,6 +666,14 @@ impl EGraph {
                 by_key.entry(&row[..inputs_len]).or_default().push(row);
             }
         }
+        // `--native-merge`: if this function is an FD-keyed constructor view, an
+        // FD conflict (same children, two eclasses) means those eclasses are
+        // CONGRUENT and must be unioned. The merge keeps the min output (the UF
+        // leader, matching union-by-min) AND injects a union of every losing
+        // eclass with the winner into the view's union-find — congruence INLINE at
+        // insert time. Collect the (a, b) pairs to enqueue after the borrow ends.
+        let native_merge_uf = self.native_merge_uf.get(&f).copied();
+        let mut union_pairs: Vec<(u32, u32)> = Vec::new();
         // Resolve each touched key; collect the rows to remove and the winner to
         // insert. Only keys with >1 candidate row can change.
         let mut new_rows: Vec<Row> = Vec::new();
@@ -682,14 +698,26 @@ impl EGraph {
             let mut winner: Vec<u32> = cands[0][..inputs_len].to_vec();
             winner.push(chosen);
             let winner: Row = winner.into_boxed_slice();
-            for row in cands {
-                if **row != *winner {
-                    drop_rows.insert((*row).clone());
+            for row in &cands {
+                if ***row != *winner {
+                    drop_rows.insert((**row).clone());
+                }
+            }
+            // Native-merge: union each candidate eclass with the chosen winner.
+            // `enqueue_union` is idempotent for already-equal ids and picks the min
+            // leader itself, so passing every candidate (incl. the winner) is safe;
+            // the dedup against `chosen` keeps the staged list small.
+            if native_merge_uf.is_some() {
+                for row in &cands {
+                    let out = row_col(row, inputs_len);
+                    if out != chosen {
+                        union_pairs.push((chosen, out));
+                    }
                 }
             }
             new_rows.push(winner);
         }
-        if drop_rows.is_empty() {
+        if drop_rows.is_empty() && union_pairs.is_empty() {
             return false;
         }
         {
@@ -712,6 +740,22 @@ impl EGraph {
                 self.index_insert_row(f, r);
             }
         }
+        // `--native-merge`: inject the congruence unions into the view's UF. The
+        // ids are drained + applied at the iteration boundary (`native_uf_drain_all`,
+        // like every other union); the displaced-id signal then drives the rebuild
+        // that re-canonicalizes any other rows holding the loser eclass. Returning
+        // `true` here keeps the outer saturate loop iterating until the unions and
+        // their rebuild fallout reach a fixpoint.
+        if let Some(uf_func) = native_merge_uf {
+            if let Some(uf) = self.native_ufs.get_mut(&uf_func) {
+                let mem = self.native_uf_members.entry(uf_func).or_default();
+                for (a, b) in &union_pairs {
+                    uf.enqueue_union(*a as i64, *b as i64);
+                    mem.insert(*a);
+                    mem.insert(*b);
+                }
+            }
+        }
         true
     }
 }
@@ -859,6 +903,17 @@ impl Backend for EGraph {
         Ok((id, canon))
     }
 
+    fn register_native_merge_view(&mut self, uf_func: FunctionId, view_func: FunctionId) {
+        // `--native-merge`: the view's eclass (OUTPUT) column belongs to `uf_func`.
+        // Record the association so `resolve_merge` injects a union into this UF on
+        // an FD conflict (same children, two eclasses) — inline congruence.
+        debug_assert!(
+            self.native_ufs.contains_key(&uf_func),
+            "register_native_merge_view: uf_func is not a native union-find"
+        );
+        self.native_merge_uf.insert(view_func, uf_func);
+    }
+
     fn table_size(&self, table: FunctionId) -> usize {
         self.mirror.get(&table).map(|s| s.len()).unwrap_or(0)
     }

egglog-experimental/src/main.rs

@@ -23,9 +23,13 @@ fn main() {
     // `--native-uf` (PR #782) so the duckdb egraph is built in native-UF mode
     // when either is set — otherwise the `--native-uf` encoding would emit
     // UF-backed functions against a relational duckdb backend.
+    // `--native-merge` does congruence inline via the backend's UnionId merge and
+    // injects the union edge into the in-core union-find, so it REQUIRES native-UF
+    // — provision the backend in native-UF mode when it is set (mirrors cli.rs's
+    // `args.native_uf = true` fold for `--native-merge`).
     let want_native_uf = argv
         .iter()
-        .any(|a| a == "--duck-native-uf" || a == "--native-uf");
+        .any(|a| a == "--duck-native-uf" || a == "--native-uf" || a == "--native-merge");
     // `--fast-rebuild` engages the duckdb backend's delta-scoped rebuild; the
     // pre-built duckdb egraph must carry it so the flag survives cli.rs's
     // short-circuit (mirror `want_native_uf`).

src/cli.rs

@@ -128,6 +128,18 @@ struct Args {
     /// Bit-exact with the full rebuild; off by default.
     #[clap(long = "fast-rebuild")]
     fast_rebuild: bool,
+    /// Native-merge: do congruence INLINE via a native backend merge instead of
+    /// the rule-encoded `@congruence_rule*` self-join. A constructor's `@<F>View`
+    /// becomes an FD view `(children) -> eclass` with a side-effecting
+    /// `MergeFn::UnionId` merge that unions the two eclasses on an FD conflict
+    /// (same children, different output) at insert time, and the separate
+    /// congruence rule is dropped from the per-iteration rebuild. This is the
+    /// non-proof half of egglog PR #933. Implies `--native-uf` + term encoding
+    /// (the injected union edge goes into the in-core union-find). `--flowlog`
+    /// only for now (the backend must inject the union on FD-conflict). Term mode
+    /// only; off by default (the rule-encoded congruence is byte-identical).
+    #[clap(long = "native-merge")]
+    native_merge: bool,
     /// Native-engine table rebuild for the term-encoding native-UF rebuild
     /// (`--nativerb`, native bridge only). Replaces the per-column
     /// `@rebuild_rule*` rebuild rules with the ENGINE's native table rebuild
@@ -162,7 +174,24 @@ pub fn cli(mut egraph: EGraph) {
         .parse_default_env()
         .init();
 
-    let args = Args::parse();
+    let mut args = Args::parse();
+
+    // `--native-merge` does congruence inline via a native backend merge that
+    // injects the union-on-FD-conflict edge into the in-core union-find — so it
+    // REQUIRES native-UF (the relational `@UF_S` parent table has no insert-time
+    // merge hook). It is implemented on the FlowLog backend only for now (the
+    // backend must inject the union on FD-conflict; other backends still map a
+    // UnionId merge to a plain min-and-drop, which would silently lose the
+    // congruence edge). Enforce flowlog and fold native-merge into native-UF so
+    // every downstream `args.native_uf` site (term encoding, backend config) is
+    // provisioned correctly.
+    if args.native_merge {
+        if !args.flowlog_backend {
+            log::error!("--native-merge is only supported with --flowlog");
+            std::process::exit(1);
+        }
+        args.native_uf = true;
+    }
 
     // The experimental backends are mutually exclusive: each swaps in a
     // different `Backend` impl for the run, so at most one may be selected.
@@ -198,20 +227,36 @@ pub fn cli(mut egraph: EGraph) {
     // encoding (enabling it here covers the case where `--term-encoding`
     // was not also passed).
     //
-    // Skip this when the caller already handed us a duckdb-backed egraph
-    // (egglog-experimental's `main` pre-builds one so its commands /
-    // primitives survive). `with_duckdb_backend` already provisions term
-    // encoding (a bridge-backed typechecker) without cloning the backend,
-    // whereas `with_term_encoding_enabled` clones `self` — and the duckdb
-    // backend's `clone_boxed` is unimplemented, so cloning it panics.
-    if (args.term_encoding || args.native_uf) && !egraph.has_duckdb_backend() {
+    // Skip this when the caller already handed us a duckdb/flowlog/feldera-backed
+    // egraph (egglog-experimental's `main` pre-builds one so its commands /
+    // primitives survive). Those `with_*_backend_config` constructors already
+    // provision term encoding (a bridge-backed typechecker) without cloning the
+    // backend, whereas `with_term_encoding_enabled` clones `self` — and those
+    // backends' `clone_boxed` is unimplemented, so cloning panics. Plain
+    // `--flowlog`/`--feldera` (no `--native-uf`/`--term-encoding`) never reach
+    // here (both flags are false); `--native-uf --flowlog` does, hence the guard.
+    if (args.term_encoding || args.native_uf)
+        && !egraph.has_duckdb_backend()
+        && !egraph.has_flowlog_backend()
+        && !egraph.has_feldera_backend()
+    {
         egraph = egraph.with_term_encoding_enabled();
     }
 
     if args.native_uf {
         egraph = egraph.with_native_uf();
     }
 
+    // `--native-merge` (set after native-UF above, which it implies): switch the
+    // encoder to the FD-keyed constructor view + inline UnionId merge. The
+    // backend egraph built below is the one that runs; for the flowlog short-
+    // circuit path the `with_native_merge` call must land on `backend_eg` too
+    // (see the experimental backend branch), so set it here on the egraph that
+    // cli ultimately runs.
+    if args.native_merge {
+        egraph = egraph.with_native_merge();
+    }
+
     // `--fast-rebuild` on the dataflow/SQL backends is wired into each backend's
     // config (`config.fast_rebuild` → `enable_fast_rebuild()`) below. On the
     // NATIVE BRIDGE (no `--duckdb`/`--feldera`/`--flowlog`) there is no separate
@@ -478,6 +523,13 @@ pub fn cli(mut egraph: EGraph) {
                 if args.native_uf && (args.flowlog_backend || args.feldera_backend) {
                     backend_eg = backend_eg.with_native_uf();
                 }
+                // `--native-merge` (flowlog only, enforced above): switch the
+                // rebuilt backend egraph's encoder to the FD-keyed view + inline
+                // UnionId merge. Must be set on `backend_eg` (the egraph that
+                // actually runs the program), not the pre-rebuild `egraph`.
+                if args.native_merge && args.flowlog_backend {
+                    backend_eg = backend_eg.with_native_merge();
+                }
                 backend_eg.parser = std::mem::take(&mut egraph.parser);
                 // Carry over the experimental user-defined commands (run-schedule,
                 // multi-extract) that egglog-experimental registered. The parser