fix(codegen): deterministic ord via single-thread fact interning

flowlog-build/src/build/pipeline.rs

@@ -52,9 +52,9 @@ impl Pipeline {
             ))
         })?;
 
-        let config = build_config(builder, program_str);
+        let mut config = build_config(builder, program_str);
         let mut program = parse(&config, &builder.include_dirs, sm)?;
-        check_program(&mut program, &config)?;
+        check_program(&mut program, &mut config)?;
         // Constant-fold after type checking (literals pinned, casts stripped)
         // and before planning, so the catalog and dataflow never see constant
         // sub-expressions.
@@ -112,5 +112,6 @@ fn build_config(builder: &Builder, program: &str) -> Config {
             .map(|p| p.to_string_lossy().into_owned())
             .collect(),
         output_to_stdout: false,
+        serialize_load: false,
     }
 }

flowlog-build/src/planner/program_planner.rs

@@ -126,7 +126,7 @@ mod tests {
         let mut sm = SourceMap::new();
         let mut program =
             Program::parse(&tmp.path().to_string_lossy(), false, &[], &mut sm).expect("parse");
-        check_program(&mut program, &Config::default()).expect("typecheck");
+        check_program(&mut program, &mut Config::default()).expect("typecheck");
         ProgramPlanner::from_program(&Config::default(), &program, &mut None).expect("plan")
     }
 

flowlog-build/tests/typechecker_errors.rs

@@ -13,7 +13,7 @@ fn typecheck(name: &str) -> (Result<(), TypeCheckError>, SourceMap) {
     let mut sm = SourceMap::new();
     let mut program = Program::parse(&fixture("typechecker", name), false, &[], &mut sm)
         .expect("fixture should parse cleanly");
-    (check_program(&mut program, &Config::default()), sm)
+    (check_program(&mut program, &mut Config::default()), sm)
 }
 
 #[test]

flowlog-common/src/config.rs

@@ -54,6 +54,10 @@ pub struct Config {
     /// Whether `.output` relations drain to stdout (`-D -`) rather than files.
     /// Derived by the CLI from `--output-dir`; always `false` in library mode.
     pub output_to_stdout: bool,
+    /// When set, fact strings are interned serially rather than in parallel, so
+    /// interning order — and therefore `ord(_)` values — is deterministic across
+    /// worker counts (`-w N` matches `-w 1`).
+    pub serialize_load: bool,
 }
 
 impl Config {
@@ -130,6 +134,11 @@ impl Config {
     pub fn output_to_stdout(&self) -> bool {
         self.output_to_stdout
     }
+
+    /// Whether fact-string interning must be serial for deterministic `ord(_)`.
+    pub fn serialize_load(&self) -> bool {
+        self.serialize_load
+    }
 }
 
 /// File stem of a program path (e.g. `galen` for `path/to/galen.dl`), or

flowlog-compiler/src/cli.rs

@@ -86,6 +86,7 @@ impl Cli {
             udf_file: self.udf_file.clone(),
             include_dirs: self.include_dirs.clone(),
             output_to_stdout: self.output_dir.as_deref() == Some("-"),
+            serialize_load: false,
         }
     }
 

flowlog-compiler/src/io/input.rs

@@ -48,7 +48,12 @@ impl Compiler {
         let has_inline_facts = !self.program.facts().is_empty();
         let needs_preload = has_file_backed_edbs || has_inline_facts;
 
-        let maybe_peers = if has_file_backed_edbs {
+        // We want a deterministic load whenever the program uses `ord`,
+        // so its value depends on interning order, deterministic across
+        // different runs; evaluation still runs on every worker.
+        let deterministic_load = self.config.serialize_load();
+
+        let maybe_peers = if has_file_backed_edbs && !deterministic_load {
             quote! { let peers = worker.peers(); }
         } else {
             quote! {}
@@ -70,9 +75,21 @@ impl Compiler {
                             .into_owned()
                     })
                     .unwrap_or_else(|| file_name);
-                quote! {
-                    rels.get_mut(#rel_name).unwrap()
-                        .apply_file(std::path::Path::new(#path), SEMIRING_ONE, peers, index);
+                if deterministic_load {
+                    // Worker 0 alone reads the whole file (`peers = 1, index = 0`);
+                    // other workers skip loading. Interning order thus matches `-w 1`.
+                    quote! {
+                        if index == 0 {
+                            rels.get_mut(#rel_name).unwrap()
+                                .apply_file(std::path::Path::new(#path), SEMIRING_ONE, 1, 0);
+                        }
+                    }
+                } else {
+                    // Each worker ingests its own ~1/N byte-range slice in parallel.
+                    quote! {
+                        rels.get_mut(#rel_name).unwrap()
+                            .apply_file(std::path::Path::new(#path), SEMIRING_ONE, peers, index);
+                    }
                 }
             })
             .collect();

flowlog-compiler/src/main.rs

@@ -23,7 +23,7 @@ fn main() {
         .init();
 
     let cli = Cli::parse();
-    let config = cli.to_config();
+    let mut config = cli.to_config();
     let options = cli.to_compile_options();
 
     // Parse the source into an AST.
@@ -37,7 +37,7 @@ fn main() {
     .unwrap_or_else(|err| emit_and_exit(err, &sm));
 
     // Type-check the program.
-    check_program(&mut program, &config).unwrap_or_else(|err| emit_and_exit(err, &sm));
+    check_program(&mut program, &mut config).unwrap_or_else(|err| emit_and_exit(err, &sm));
 
     // Constant-fold after type checking, before planning.
     fold_constants(&mut program);

flowlog-typechecker/src/lib.rs

@@ -78,11 +78,11 @@ use crate::env::PrimitiveEnv;
 /// Runs two passes: Pass 1 ([`primitive`]) checks primitive `DataType`s and
 /// pins literals; Pass 2 ([`subtype`]) enforces subtype rules and lowers
 /// `as()` casts. `config` is consulted for config-gated builtins.
-pub fn check_program(program: &mut Program, config: &Config) -> Result<(), TypeCheckError> {
+pub fn check_program(program: &mut Program, config: &mut Config) -> Result<(), TypeCheckError> {
     let env = PrimitiveEnv::from_program(program);
 
     primitive::rule::check_and_pin_rules(program, &env)?;
-    primitive::check_builtin_config_requirements(program, config)?;
+    primitive::builtin::check_ord(program, config)?;
     primitive::fact::check_and_pin_facts(program.facts_mut(), &env)?;
 
     subtype::check_and_lower(program)

flowlog-typechecker/src/primitive/builtin.rs

@@ -0,0 +1,65 @@
+//! Builtin checks.
+
+use flowlog_common::Config;
+use flowlog_common::Span;
+use flowlog_parser::Arithmetic;
+use flowlog_parser::BuiltinOperator;
+use flowlog_parser::Factor;
+use flowlog_parser::HeadArg;
+use flowlog_parser::Predicate;
+use flowlog_parser::Program;
+
+use crate::TypeCheckError;
+
+/// Check `ord(_)` usage in one walk over the program (including `loop`/`fixpoint`
+/// bodies):
+///
+/// - reject `ord` without `--str-intern` (`OrdRequiresStrIntern`);
+/// - otherwise set [`Config::serialize_load`] so the loader interns serially,
+///   keeping `ord` deterministic across worker counts.
+pub(crate) fn check_ord(program: &Program, config: &mut Config) -> Result<(), TypeCheckError> {
+    fn arith(a: &Arithmetic) -> Option<Span> {
+        factor(a.init()).or_else(|| a.rest().iter().find_map(|(_, f)| factor(f)))
+    }
+    fn factor(f: &Factor) -> Option<Span> {
+        match f {
+            Factor::Var(_) | Factor::Const(_) => None,
+            Factor::FnCall(fc) => fc.args().iter().find_map(arith),
+            Factor::Builtin(bc) if bc.op() == BuiltinOperator::Ord => Some(bc.span()),
+            Factor::Builtin(bc) => bc.args().iter().find_map(arith),
+            Factor::Cast(c) => factor(c.inner()),
+            Factor::Group(a) => arith(a),
+            Factor::Tuple(t) => t.exprs().find_map(arith),
+            Factor::TupleProj { tuple, .. } => arith(tuple),
+        }
+    }
+
+    // Chain `as_loop()`: `as_rules()` alone skips rules nested in
+    // `loop`/`fixpoint` blocks, so an `ord` there would otherwise escape both
+    // the error and the serial-load decision.
+    let ord_span = program.segments().iter().find_map(|seg| {
+        let plain = seg.as_rules().iter();
+        let looped = seg.as_loop().into_iter().flat_map(|b| b.rules().iter());
+        plain.chain(looped).find_map(|rule| {
+            let body = rule.rhs().iter().find_map(|p| match p {
+                Predicate::PositiveAtom(_) | Predicate::NegativeAtom(_) => None,
+                Predicate::Compare(c) => arith(c.left()).or_else(|| arith(c.right())),
+            });
+            body.or_else(|| {
+                rule.head().head_arguments().iter().find_map(|h| match h {
+                    HeadArg::Var(_) => None,
+                    HeadArg::Arith(a) => arith(a),
+                    HeadArg::Aggregation(agg) => arith(agg.arithmetic()),
+                })
+            })
+        })
+    });
+
+    if let Some(span) = ord_span {
+        if !config.str_intern_enabled() {
+            return Err(TypeCheckError::OrdRequiresStrIntern { span });
+        }
+        config.serialize_load = true;
+    }
+    Ok(())
+}

flowlog-typechecker/src/primitive/mod.rs

@@ -5,9 +5,11 @@
 //! pins expressions; `atom`, `fact`, `compare`, and `rule` (the composer)
 //! check and pin each construct. Names read `<verb>_<node>` — `infer`, `check`,
 //! `pin`, or `check_and_pin` for both. `check_program` calls
-//! `rule::check_and_pin_rules` and `fact::check_and_pin_facts` directly.
+//! `rule::check_and_pin_rules`, `builtin::check_ord`, and
+//! `fact::check_and_pin_facts` directly.
 
 mod atom;
+pub(crate) mod builtin;
 mod compare;
 mod expr;
 pub(crate) mod fact;
@@ -16,74 +18,10 @@ mod ty;
 
 use std::collections::HashMap;
 
-use flowlog_common::Config;
 use flowlog_common::Span;
-use flowlog_parser::Arithmetic;
-use flowlog_parser::BuiltinOperator;
 use flowlog_parser::DataType;
-use flowlog_parser::Factor;
-use flowlog_parser::HeadArg;
-use flowlog_parser::Predicate;
-use flowlog_parser::Program;
-
-use crate::TypeCheckError;
 
 /// Var -> (first-seen type, first-seen span). Later uses must agree.
 /// Private, so it stays confined to the `primitive` subtree — the subtype
 /// pass keeps its own `TypeId`-keyed map.
 type Bindings = HashMap<String, (DataType, Span)>;
-
-/// Reject built-in calls whose semantics depend on a build flag that
-/// isn't enabled — today only `ord(_)`, which needs `--str-intern`.
-pub(crate) fn check_builtin_config_requirements(
-    program: &Program,
-    config: &Config,
-) -> Result<(), TypeCheckError> {
-    if config.str_intern_enabled() {
-        return Ok(());
-    }
-    fn check_arith(a: &Arithmetic) -> Result<(), TypeCheckError> {
-        check_factor(a.init())?;
-        for (_, f) in a.rest() {
-            check_factor(f)?;
-        }
-        Ok(())
-    }
-    fn check_factor(f: &Factor) -> Result<(), TypeCheckError> {
-        match f {
-            Factor::Var(_) | Factor::Const(_) => Ok(()),
-            Factor::FnCall(fc) => fc.args().iter().try_for_each(check_arith),
-            Factor::Builtin(bc) => {
-                if bc.op() == BuiltinOperator::Ord {
-                    return Err(TypeCheckError::OrdRequiresStrIntern { span: bc.span() });
-                }
-                bc.args().iter().try_for_each(check_arith)
-            }
-            Factor::Cast(c) => check_factor(c.inner()),
-            Factor::Group(a) => check_arith(a),
-            Factor::Tuple(r) => r.exprs().try_for_each(check_arith),
-            Factor::TupleProj { tuple, .. } => check_arith(tuple),
-        }
-    }
-    for segment in program.segments() {
-        for rule in segment.as_rules() {
-            for predicate in rule.rhs() {
-                match predicate {
-                    Predicate::PositiveAtom(_) | Predicate::NegativeAtom(_) => {}
-                    Predicate::Compare(cmp) => {
-                        check_arith(cmp.left())?;
-                        check_arith(cmp.right())?;
-                    }
-                }
-            }
-            for head_arg in rule.head().head_arguments() {
-                match head_arg {
-                    HeadArg::Var(_) => {}
-                    HeadArg::Arith(a) => check_arith(a)?,
-                    HeadArg::Aggregation(agg) => check_arith(agg.arithmetic())?,
-                }
-            }
-        }
-    }
-    Ok(())
-}

flowlog-typechecker/tests/common.rs

@@ -29,6 +29,6 @@ pub fn parse_program(src: &str) -> Result<Program, ParseError> {
 pub fn parse_and_check_result(src: &str) -> Result<Program, TypeCheckError> {
     let mut program =
         parse_program(src).expect("parse should succeed; this test exercises typecheck only");
-    check_program(&mut program, &Config::default())?;
+    check_program(&mut program, &mut Config::default())?;
     Ok(program)
 }