From bfad4d33e8965c6727f0735b514cbd29f8e890b8 Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Tue, 23 Jun 2026 19:35:17 -0700 Subject: [PATCH 1/7] feat(sqllib): pin now() via FELDERA_NOW_OVERRIDE_MS for deterministic testing now() honors the FELDERA_NOW_OVERRIDE_MS env var (milliseconds since the Unix epoch, read once into a LazyLock) and returns that fixed instant instead of the wall clock, so a harvest or test gets a reproducible now(). Factored into now_from(Option) so the override path is unit-tested without env-var flakiness. Signed-off-by: Gerd Zellweger --- crates/sqllib/src/timestamp.rs | 30 +++++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/crates/sqllib/src/timestamp.rs b/crates/sqllib/src/timestamp.rs index 6f9ed1917f2..a09151a0641 100644 --- a/crates/sqllib/src/timestamp.rs +++ b/crates/sqllib/src/timestamp.rs @@ -756,6 +756,17 @@ mod tests { use super::*; use chrono::FixedOffset; + #[test] + fn test_now_override() { + // A pinned override returns that exact instant; no override reads the wall clock. + assert_eq!( + now_from(Some(1_577_836_800_000)).milliseconds(), + 1_577_836_800_000 + ); + let wall = now_from(None).milliseconds(); + assert!(wall > 1_577_836_800_000, "wall clock is past the year 2020"); + } + #[test] fn test_parse_tstz() { // Check that the parsing from string accepts the same formats @@ -891,9 +902,26 @@ pub fn convert_timezone___(src_tz: Zone, dst_tz: Zone, ts: Timestamp) -> Option< some_nullable_function3!(convert_timezone, Zone, Zone, Timestamp, Timestamp); +/// A fixed `now()` for deterministic harvests and tests: when `FELDERA_NOW_OVERRIDE_MS` holds an +/// integer count of milliseconds since the Unix epoch, `now()` returns that instant instead of the +/// wall clock, so every call in a run agrees. Read once at first use. +static NOW_OVERRIDE_MS: std::sync::LazyLock> = std::sync::LazyLock::new(|| { + std::env::var("FELDERA_NOW_OVERRIDE_MS") + .ok() + .and_then(|ms| ms.parse::().ok()) +}); + +/// Resolve `now()` from an optional override: the pinned instant when set, the wall clock otherwise. +fn now_from(override_ms: Option) -> Timestamp { + match override_ms { + Some(ms) => Timestamp::from_milliseconds(ms), + None => Timestamp::from_dateTime(Utc::now()), + } +} + #[doc(hidden)] pub fn now() -> Timestamp { - Timestamp::from_dateTime(Utc::now()) + now_from(*NOW_OVERRIDE_MS) } #[doc(hidden)] From 4e44f2446b4976c4a090647e14b8c97eaceb50ad Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Tue, 23 Jun 2026 19:35:59 -0700 Subject: [PATCH 2/7] feat(sql-compiler): expression oracle harvest backend ExpressionOracleHarvest extracts each operator closure from a compiled circuit and emits it as an {ir, inputs, outputs} record: the ground-truth corpus the forge circuit-expr interpreter is validated against. BaseSQLTests wires it into the test path. Signed-off-by: Gerd Zellweger --- .../backend/ExpressionOracleHarvest.java | 356 ++++++++++++++++++ .../compiler/sql/tools/BaseSQLTests.java | 2 + 2 files changed, 358 insertions(+) create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java new file mode 100644 index 00000000000..d6000f15b1c --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java @@ -0,0 +1,356 @@ +package org.dbsp.sqlCompiler.compiler.backend; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.dbsp.sqlCompiler.circuit.DBSPCircuit; +import org.dbsp.sqlCompiler.circuit.DBSPDeclaration; +import org.dbsp.sqlCompiler.circuit.operator.DBSPFilterOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPJoinOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPLeftJoinOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPMapIndexOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPMapOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPStreamJoinOperator; +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.backend.rust.ToRustInnerVisitor; +import org.dbsp.sqlCompiler.ir.IDBSPInnerNode; +import org.dbsp.sqlCompiler.ir.expression.DBSPClosureExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; +import org.dbsp.sqlCompiler.ir.statement.DBSPStaticItem; +import org.dbsp.sqlCompiler.ir.type.DBSPType; +import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeRawTuple; +import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeRef; +import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeTuple; +import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeTupleBase; +import org.dbsp.util.IndentStreamBuilder; +import org.dbsp.util.JsonStream; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Harvests stateless scalar closures from compiled test circuits into per-closure + * JSON records, the raw material for the standalone expression oracle. + * + *

Gated on the {@code FELDERA_EXPR_ORACLE_DIR} environment variable, so it runs + * only when explicitly collecting a corpus and never perturbs a normal test run. + * Every failure is swallowed: harvesting must not break the test that triggered it. + * + *

Harvested shapes: + *

    + *
  • {@code Map}/{@code Filter} (one parameter, {@code &Tup}): a projection + * or predicate over one row;
    • + *
  • equi-join projections ({@code Join}/{@code StreamJoin}/{@code LeftJoin}, three + * parameters {@code &key, &left, &right}, each a {@code &Tup}).
    • + *
+ * Every leaf column must be a scalar the oracle runtime can sample, and each tuple at + * most ten wide. Closures may reference interned-string and decimal STATIC constants; + * the referenced declarations travel with the record so the generated crate can emit + * them. Indexed (raw-tuple) parameters and unsupported result shapes are skipped, so + * the generated crate always compiles. + */ +public final class ExpressionOracleHarvest { + private ExpressionOracleHarvest() {} + + /** Rust leaf types the oracle runtime implements `Sample` + `ToOracleJson` for. */ + private static final Set SUPPORTED_LEAF = Set.of( + "bool", "i8", "i16", "i32", "i64", "i128", "u8", "u16", "u32", "u64", "u128", "F32", + "F64", "SqlString", "Date", "Time", "Timestamp", "TimestampTz", "ShortInterval", + "LongInterval", "ByteArray", "Uuid", "Variant", "GeoPoint"); + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + // Coverage counters, accumulated across the suite (surefire reuses one JVM) and + // rewritten to `_stats.json` after every circuit so the last write is the total. + private static final AtomicInteger candidates = new AtomicInteger(); + private static final AtomicInteger emitted = new AtomicInteger(); + private static final Map skipped = new ConcurrentHashMap<>(); + + private static void skip(String reason) { + skipped.computeIfAbsent(reason, ignored -> new AtomicInteger()).incrementAndGet(); + } + + public static void maybeHarvest(DBSPCircuit circuit, DBSPCompiler compiler) { + String dir = System.getenv("FELDERA_EXPR_ORACLE_DIR"); + if (dir == null || dir.isBlank()) { + return; + } + try { + harvest(circuit, compiler, Paths.get(dir)); + } catch (Throwable ignored) { + // Harvesting is best-effort; a normal test run must not depend on it. + } + } + + private static void harvest(DBSPCircuit circuit, DBSPCompiler compiler, Path dir) + throws Exception { + Files.createDirectories(dir); + + for (DBSPOperator operator : circuit.getAllOperators()) { + if (!(operator instanceof DBSPSimpleOperator simple) || !isHarvestable(operator)) { + continue; + } + DBSPExpression function = simple.function; + if (!(function instanceof DBSPClosureExpression closure)) { + continue; + } + int arity = closure.parameters.length; + if (arity != 1 && arity != 3) { + continue; + } + candidates.incrementAndGet(); + // A closure over an indexed input takes a borrowed key/value pair + // `&(&key, &value)`; everything else takes plain `&Tup` rows. + boolean index = arity == 1 && isIndexedParam(closure.parameters[0].getType()); + ArrayNode params = index ? indexParams(compiler, closure) : scalarParams(compiler, closure); + if (params == null) { + skip("unsupported_parameter"); + continue; + } + if (!isSupportedResult(compiler, closure.getResultType())) { + skip("unsupported_result"); + continue; + } + + String rustClosure = ToRustInnerVisitor.toRustString(compiler, closure, null, false); + List declarations = referencedDeclarations(circuit, rustClosure); + if (declarations == null) { + // References a declaration that is not a STATIC constant (a function or + // struct); emitting those is out of scope. + skip("non_static_declaration"); + continue; + } + + String hash = sha1(rustClosure); + ObjectNode record = MAPPER.createObjectNode(); + record.put("name", "case_" + hash); + record.put("operator", operator.getClass().getSimpleName()); + record.put("index", index); + record.set("params", params); + attachStatics(compiler, declarations, record); + record.put("rust_closure", rustClosure); + record.set("ir", buildIr(compiler, closure, declarations)); + + // Filename is the dedup key: identical closures from different tests and + // JVMs converge on one file rather than racing an append. + MAPPER.writeValue(dir.resolve(hash + ".json").toFile(), record); + emitted.incrementAndGet(); + } + writeStats(dir); + } + + /** + * Coverage so far, rewritten after every circuit. `candidates` counts the + * stateless scalar closures of a harvestable shape; `emitted` counts those that + * passed every filter (before dedup); the unique total is the record-file count. + */ + private static void writeStats(Path dir) throws Exception { + ObjectNode stats = MAPPER.createObjectNode(); + stats.put("candidates", candidates.get()); + stats.put("emitted", emitted.get()); + ObjectNode bySkip = stats.putObject("skipped"); + skipped.forEach((reason, count) -> bySkip.put(reason, count.get())); + MAPPER.writeValue(dir.resolve("_stats.json").toFile(), stats); + } + + /** The operator kinds whose closure is a stateless scalar projection or predicate. */ + private static boolean isHarvestable(DBSPOperator operator) { + return operator instanceof DBSPMapOperator + || operator instanceof DBSPMapIndexOperator + || operator instanceof DBSPFilterOperator + || operator instanceof DBSPJoinOperator + || operator instanceof DBSPStreamJoinOperator + || operator instanceof DBSPLeftJoinOperator; + } + + /** + * One entry per closure parameter, each `&Tup`; null if any parameter + * is not that shape or a leaf is not sampleable. + */ + private static ArrayNode scalarParams(DBSPCompiler compiler, DBSPClosureExpression closure) { + ArrayNode params = MAPPER.createArrayNode(); + for (var parameter : closure.parameters) { + if (!(parameter.getType() instanceof DBSPTypeRef ref) + || !(ref.type instanceof DBSPTypeTuple tuple)) { + return null; + } + ObjectNode param = tupleParam(compiler, tuple, true); + if (param == null) { + return null; + } + params.add(param); + } + return params; + } + + /** Whether the parameter is a borrowed key/value pair `&(&keyTuple, &valueTuple)`. */ + private static boolean isIndexedParam(DBSPType type) { + return type instanceof DBSPTypeRef ref + && ref.type instanceof DBSPTypeRawTuple pair + && pair.tupFields.length == 2; + } + + /** + * A closure over an indexed input takes one parameter, a borrowed key/value pair + * {@code &(&keyTuple, &valueTuple)}. The two tuples become the parameter list. + */ + private static ArrayNode indexParams(DBSPCompiler compiler, DBSPClosureExpression closure) { + if (!(closure.parameters[0].getType() instanceof DBSPTypeRef ref) + || !(ref.type instanceof DBSPTypeRawTuple pair) + || pair.tupFields.length != 2) { + return null; + } + ArrayNode params = MAPPER.createArrayNode(); + for (DBSPType field : pair.tupFields) { + if (!(field instanceof DBSPTypeRef inner) + || !(inner.type instanceof DBSPTypeTuple tuple)) { + return null; + } + // A nullable key or value tuple would need Option construction the index + // driver does not do; skip it. + ObjectNode param = tupleParam(compiler, tuple, false); + if (param == null) { + return null; + } + params.add(param); + } + return params; + } + + /** A parameter descriptor for one tuple of scalar leaves, or null if unsupported. */ + private static ObjectNode tupleParam(DBSPCompiler compiler, DBSPTypeTuple tuple, boolean allowNull) { + if (tuple.mayBeNull && !allowNull) { + return null; + } + DBSPType[] fields = tuple.tupFields; + if (fields.length < 1 || fields.length > 10) { + return null; + } + ObjectNode param = MAPPER.createObjectNode(); + // A nullable parameter tuple is a left join's absent right row (`&Option`); + // the generator samples it as Some/None. + param.put("nullable", tuple.mayBeNull); + ArrayNode leaves = param.putArray("leaves"); + for (DBSPType field : fields) { + String rust = ToRustInnerVisitor.toRustString(compiler, field, null, false); + if (!isSupportedLeaf(rust)) { + return null; + } + leaves.add(rust); + } + return param; + } + + private static boolean isSupportedLeaf(String rustType) { + String base = stripOption(rustType); + return SUPPORTED_LEAF.contains(base) || base.startsWith("SqlDecimal"); + } + + /** + * The result type the generated crate will JSON-encode: a scalar (predicate), or a + * tuple / raw tuple of supported values (a projection or an indexed key/value pair). + */ + private static boolean isSupportedResult(DBSPCompiler compiler, DBSPType type) { + if (type instanceof DBSPTypeRef ref) { + return isSupportedResult(compiler, ref.type); + } + if (type instanceof DBSPTypeTupleBase tuple) { + for (DBSPType field : tuple.tupFields) { + if (!isSupportedResult(compiler, field)) { + return false; + } + } + return true; + } + return isSupportedLeaf(ToRustInnerVisitor.toRustString(compiler, type, null, false)); + } + + private static String stripOption(String rustType) { + if (rustType.startsWith("Option<") && rustType.endsWith(">")) { + return rustType.substring("Option<".length(), rustType.length() - 1).trim(); + } + return rustType; + } + + /** + * The circuit declarations the closure references (by name). Null if it references a + * declaration that is not a STATIC constant, which the generator cannot emit. + */ + private static List referencedDeclarations( + DBSPCircuit circuit, String rustClosure) { + List referenced = new ArrayList<>(); + for (DBSPDeclaration declaration : circuit.declarations) { + String name = declaration.getName(); + if (name.isBlank() || !rustClosure.contains(name)) { + continue; + } + if (!(declaration.item instanceof DBSPStaticItem)) { + return null; + } + referenced.add(declaration); + } + return referenced; + } + + /** + * Render each referenced STATIC as a function-local declaration plus its initializer. + * The generated crate emits all declarations first, then all initializers (init is + * idempotent and the value is computed lazily), then the closure that reads them. + */ + private static void attachStatics( + DBSPCompiler compiler, List declarations, ObjectNode record) { + ArrayNode decls = record.putArray("static_decls"); + ArrayNode inits = record.putArray("static_inits"); + for (DBSPDeclaration declaration : declarations) { + var stat = ((DBSPStaticItem) declaration.item).expression; + String name = stat.getName(); + String type = ToRustInnerVisitor.toRustString(compiler, stat.getType(), null, false); + String init = ToRustInnerVisitor.toRustString(compiler, stat.initializer, null, false); + decls.add("static " + name + ": StaticLazy<" + type + "> = StaticLazy::new();"); + inits.add(name + ".init(move || " + init + ");"); + } + } + + /** The closure IR plus the STATIC declarations it references, self-contained. */ + private static ObjectNode buildIr( + DBSPCompiler compiler, DBSPClosureExpression closure, List declarations) + throws Exception { + ArrayNode irDeclarations = MAPPER.createArrayNode(); + for (DBSPDeclaration declaration : declarations) { + ObjectNode wrapper = MAPPER.createObjectNode(); + wrapper.set("item", MAPPER.readTree(innerJson(compiler, declaration.item))); + irDeclarations.add(wrapper); + } + ObjectNode ir = MAPPER.createObjectNode(); + ir.set("declarations", irDeclarations); + ir.set("function", MAPPER.readTree(innerJson(compiler, closure))); + return ir; + } + + private static String innerJson(DBSPCompiler compiler, IDBSPInnerNode node) { + JsonStream stream = new JsonStream(new IndentStreamBuilder()); + ToJsonInnerVisitor visitor = new ToJsonInnerVisitor(compiler, stream, 1); + node.accept(visitor); + return visitor.getJsonString(); + } + + private static String sha1(String value) throws Exception { + byte[] digest = MessageDigest.getInstance("SHA-1").digest(value.getBytes("UTF-8")); + StringBuilder hex = new StringBuilder(digest.length * 2); + for (byte b : digest) { + hex.append(Character.forDigit((b >> 4) & 0xf, 16)); + hex.append(Character.forDigit(b & 0xf, 16)); + } + return hex.toString(); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java index 4b57fffb5e5..9283e6669a3 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java @@ -24,6 +24,7 @@ package org.dbsp.sqlCompiler.compiler.sql.tools; import org.dbsp.sqlCompiler.circuit.DBSPCircuit; +import org.dbsp.sqlCompiler.compiler.backend.ExpressionOracleHarvest; import org.dbsp.sqlCompiler.compiler.CompilerOptions; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; import org.dbsp.sqlCompiler.compiler.backend.rust.RustFileWriter; @@ -431,6 +432,7 @@ public static DBSPCircuit getCircuit(DBSPCompiler compiler) { compiler.showErrors(System.err); throw new RuntimeException("No circuit produced"); } + ExpressionOracleHarvest.maybeHarvest(circuit, compiler); return circuit; } From 0587c76ad8c24c848f16db54814420a9457f0dcc Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Tue, 23 Jun 2026 19:36:14 -0700 Subject: [PATCH 3/7] wip(pipeline-manager): run the crucible engine Compile and launch the crucible single-binary engine from the pipeline manager: the rust compiler and local runner build and run the crucible binary, with the program and SQL-compiler plumbing carrying the engine selection. Signed-off-by: Gerd Zellweger --- crates/pipeline-manager/src/compiler.rs | 2 +- .../src/compiler/rust_compiler.rs | 280 ++++++++++-------- .../src/compiler/sql_compiler.rs | 6 +- .../pipeline-manager/src/db/types/program.rs | 22 +- .../src/runner/local_runner.rs | 35 ++- 5 files changed, 221 insertions(+), 124 deletions(-) diff --git a/crates/pipeline-manager/src/compiler.rs b/crates/pipeline-manager/src/compiler.rs index a6bcaed3d35..eb9f1e9d144 100644 --- a/crates/pipeline-manager/src/compiler.rs +++ b/crates/pipeline-manager/src/compiler.rs @@ -1,6 +1,6 @@ pub mod error; pub mod main; -mod rust_compiler; +pub(crate) mod rust_compiler; mod sql_compiler; #[cfg(test)] mod test; diff --git a/crates/pipeline-manager/src/compiler/rust_compiler.rs b/crates/pipeline-manager/src/compiler/rust_compiler.rs index ea6de2232d6..2bea83fd07e 100644 --- a/crates/pipeline-manager/src/compiler/rust_compiler.rs +++ b/crates/pipeline-manager/src/compiler/rust_compiler.rs @@ -900,6 +900,12 @@ pub(super) struct RustCompilationResult { /// Returns the program binary URL, source checksum, integrity checksum, /// duration, and compilation information. #[allow(clippy::too_many_arguments)] +/// HACK (crucible prototype): content of the placeholder delivered instead of +/// a pipeline binary when `runtime_version: "crucible"`. The local runner +/// sniffs this prefix on the fetched binary and execs the crucible executor. +pub const CRUCIBLE_PLACEHOLDER_MARKER: &[u8] = + b"crucible placeholder binary; the runner execs the crucible executor\n"; + pub async fn perform_rust_compilation( common_config: &CommonConfig, config: &CompilerConfig, @@ -1224,6 +1230,11 @@ pub async fn resolve_runtime_sha( match runtime_version { RuntimeSelector::Sha(sha) => Ok(sha.clone()), RuntimeSelector::Platform(platform_sha) => Ok(platform_sha.clone()), + // The crucible placeholder branch skips cargo, which is the only + // caller needing a revision SHA. + RuntimeSelector::Crucible => { + unreachable!("crucible compiles no pipeline binary") + } RuntimeSelector::Version(version) => { let repo_location = runtime_version.runtime_sources(config); match Command::new("git") @@ -1442,7 +1453,9 @@ async fn prepare_workspace( // --------------------- // Make sure the runtime version is checked out. let runtime_sources = requested_runtime_version.runtime_sources(config); - if !requested_runtime_version.is_platform() { + // Crucible uses the platform sources (no pipeline binary is built), so no + // checkout applies. + if !requested_runtime_version.is_platform() && !requested_runtime_version.is_crucible() { let repo_location = PathBuf::from(&runtime_sources); checkout_runtime_version( &repo_location, @@ -1461,7 +1474,7 @@ async fn prepare_workspace( // --------------------- // Contains all the (indirect and direct) dependencies of the crates besides UDF. // The original is copied over each time such that the starting point is the same. - if requested_runtime_version.is_platform() { + if requested_runtime_version.is_platform() || requested_runtime_version.is_crucible() { let cargo_lock_source_path = Path::new(&config.compilation_cargo_lock_path); let cargo_lock_target_path = workspace_dir.join("Cargo.lock"); copy_file(cargo_lock_source_path, &cargo_lock_target_path).await?; @@ -1545,142 +1558,171 @@ async fn call_compiler( ))?; let optional_env_rustflags = std::env::var_os("RUSTFLAGS"); - // Formulate command - let mut command = Command::new("cargo"); + let engine_is_crucible = runtime_selector.is_crucible(); - // get env vars that start with SCCACHE - let preserved_env_vars: Vec<(String, String)> = std::env::vars() - .filter(|(key, _)| key.starts_with("SCCACHE")) - .collect(); + // Short-circuit (crucible prototype): with `runtime_version: "crucible"`, + // crucible is the executor and the compiled binary is never run. Skip the + // slow cargo build and stand in a marker placeholder at the path cargo + // would produce; the bookkeeping below checksums and delivers it and marks + // the program compiled. The runner detects the marker and execs the + // crucible binary, not this file. Remove once crucible is the standard + // executor. + let compilation_info = if engine_is_crucible { + let source_file_path = workspace_dir + .join("target") + .join(profile.to_target_folder()) + .join(crate_name_pipeline_main(pipeline_id)); + if let Some(parent) = source_file_path.parent() { + std::fs::create_dir_all(parent).map_err(|e| { + RustCompilationError::SystemError(format!("creating target dir: {e}")) + })?; + } + std::fs::write(&source_file_path, CRUCIBLE_PLACEHOLDER_MARKER).map_err(|e| { + RustCompilationError::SystemError(format!("writing placeholder binary: {e}")) + })?; + RustCompilationInfo { + exit_code: 0, + stdout: "rustc skipped: runtime_version is 'crucible'".to_string(), + stderr: String::new(), + } + } else { + // Formulate command + let mut command = Command::new("cargo"); - command.env_clear(); - command.env("PATH", env_path); - if !runtime_selector.is_platform() { - command.env( - "FELDERA_RUNTIME_OVERRIDE", - resolve_runtime_sha(runtime_selector, config).await?, - ); - } - if let Some(env_rustflags) = optional_env_rustflags { - command.env("RUSTFLAGS", env_rustflags); - } + // get env vars that start with SCCACHE + let preserved_env_vars: Vec<(String, String)> = std::env::vars() + .filter(|(key, _)| key.starts_with("SCCACHE")) + .collect(); - if let Some(rustc_wrapper) = std::env::var_os("RUSTC_WRAPPER") { - command.env("RUSTC_WRAPPER", rustc_wrapper); - } + command.env_clear(); + command.env("PATH", env_path); + if !runtime_selector.is_platform() { + command.env( + "FELDERA_RUNTIME_OVERRIDE", + resolve_runtime_sha(runtime_selector, config).await?, + ); + } + if let Some(env_rustflags) = optional_env_rustflags { + command.env("RUSTFLAGS", env_rustflags); + } - // Preserve CARGO_INCREMENTAL if set, to allow sccache to work properly. - if let Some(cargo_incremental) = std::env::var_os("CARGO_INCREMENTAL") { - command.env("CARGO_INCREMENTAL", cargo_incremental); - } + if let Some(rustc_wrapper) = std::env::var_os("RUSTC_WRAPPER") { + command.env("RUSTC_WRAPPER", rustc_wrapper); + } - // Preserve AWS_PROFILE if set, to allow sccache to use - // credentials from there. - // we avoid passing all AWS_* env vars to prevent leaking - // credentials from the malicious build scripts. - if let Some(aws_profile) = std::env::var_os("AWS_PROFILE") { - command.env("AWS_PROFILE", aws_profile); - } + // Preserve CARGO_INCREMENTAL if set, to allow sccache to work properly. + if let Some(cargo_incremental) = std::env::var_os("CARGO_INCREMENTAL") { + command.env("CARGO_INCREMENTAL", cargo_incremental); + } - for (key, value) in preserved_env_vars { - command.env(key, value); - } + // Preserve AWS_PROFILE if set, to allow sccache to use + // credentials from there. + // we avoid passing all AWS_* env vars to prevent leaking + // credentials from the malicious build scripts. + if let Some(aws_profile) = std::env::var_os("AWS_PROFILE") { + command.env("AWS_PROFILE", aws_profile); + } - command - // Set compiler stack size to 20MB (10x the default) to prevent - // SIGSEGV when the compiler runs out of stack on large programs. - .env("RUST_MIN_STACK", "20971520") - .current_dir(&workspace_dir) - .arg("build") - .arg("--workspace") - .arg("--profile") - .arg(profile.to_string()) - .stdin(Stdio::null()) - .stdout(Stdio::from(stdout_file.into_std().await)) - .stderr(Stdio::from(stderr_file.into_std().await)) - // Setting it to zero sets the process group ID to the PID. - // This is done to be able to kill any subprocesses that are spawned. - .process_group(0); - - // Start process - let mut process = command.spawn().map_err(|e| { - RustCompilationError::SystemError( - UtilError::IoError("running 'cargo build'".to_string(), e).to_string(), - ) - })?; + for (key, value) in preserved_env_vars { + command.env(key, value); + } - // Retrieve process group ID and create a terminator - // which ends the group when going out of scope. - let Some(process_group) = process.id() else { - return Err(RustCompilationError::SystemError( - "unable to retrieve pid".to_string(), - )); - }; - let mut terminator = ProcessGroupTerminator::new("Rust compilation", process_group); - - // Wait for process to exit while regularly checking if the pipeline still exists - // and has not had its program get updated - let exit_status = loop { - match process.try_wait() { - Ok(exit_status) => match exit_status { - None => { - if let Some(db) = db.clone() { - match db - .lock() - .await - .get_pipeline_by_id_for_monitoring(tenant_id, pipeline_id) - .await - { - Ok(pipeline) => { - if pipeline.program_version != program_version { - return Err(RustCompilationError::Outdated); + command + // Set compiler stack size to 20MB (10x the default) to prevent + // SIGSEGV when the compiler runs out of stack on large programs. + .env("RUST_MIN_STACK", "20971520") + .current_dir(&workspace_dir) + .arg("build") + .arg("--workspace") + .arg("--profile") + .arg(profile.to_string()) + .stdin(Stdio::null()) + .stdout(Stdio::from(stdout_file.into_std().await)) + .stderr(Stdio::from(stderr_file.into_std().await)) + // Setting it to zero sets the process group ID to the PID. + // This is done to be able to kill any subprocesses that are spawned. + .process_group(0); + + // Start process + let mut process = command.spawn().map_err(|e| { + RustCompilationError::SystemError( + UtilError::IoError("running 'cargo build'".to_string(), e).to_string(), + ) + })?; + + // Retrieve process group ID and create a terminator + // which ends the group when going out of scope. + let Some(process_group) = process.id() else { + return Err(RustCompilationError::SystemError( + "unable to retrieve pid".to_string(), + )); + }; + let mut terminator = ProcessGroupTerminator::new("Rust compilation", process_group); + + // Wait for process to exit while regularly checking if the pipeline still exists + // and has not had its program get updated + let exit_status = loop { + match process.try_wait() { + Ok(exit_status) => match exit_status { + None => { + if let Some(db) = db.clone() { + match db + .lock() + .await + .get_pipeline_by_id_for_monitoring(tenant_id, pipeline_id) + .await + { + Ok(pipeline) => { + if pipeline.program_version != program_version { + return Err(RustCompilationError::Outdated); + } + } + Err(DBError::UnknownPipeline { .. }) => { + return Err(RustCompilationError::NoLongerExists); + } + Err(e) => { + error!( + pipeline_id = %pipeline_id, + pipeline = pipeline_name_for_log.as_str(), + "Rust compilation outdated check failed due to database error: {e}" + ) + // As preemption check failing is not fatal, compilation will continue } - } - Err(DBError::UnknownPipeline { .. }) => { - return Err(RustCompilationError::NoLongerExists); - } - Err(e) => { - error!( - pipeline_id = %pipeline_id, - pipeline = pipeline_name_for_log.as_str(), - "Rust compilation outdated check failed due to database error: {e}" - ) - // As preemption check failing is not fatal, compilation will continue } } } + Some(exit_status) => break exit_status, + }, + Err(e) => { + return Err(RustCompilationError::SystemError( + UtilError::IoError("waiting for 'cargo build'".to_string(), e).to_string(), + )); } - Some(exit_status) => break exit_status, - }, - Err(e) => { - return Err(RustCompilationError::SystemError( - UtilError::IoError("waiting for 'cargo build'".to_string(), e).to_string(), - )); } - } - sleep(COMPILATION_CHECK_INTERVAL).await; - }; + sleep(COMPILATION_CHECK_INTERVAL).await; + }; - // Once the process has exited, it is no longer needed to terminate its process group - terminator.cancel(); + // Once the process has exited, it is no longer needed to terminate its process group + terminator.cancel(); - // Check presence of exit status code - let Some(exit_code) = exit_status.code() else { - // No exit status code present because the process was terminated by a signal - return Err(RustCompilationError::TerminatedBySignal); - }; + // Check presence of exit status code + let Some(exit_code) = exit_status.code() else { + // No exit status code present because the process was terminated by a signal + return Err(RustCompilationError::TerminatedBySignal); + }; - // Read stdout and stderr - let stdout = read_file_content(&stdout_file_path).await?; - let stderr = read_file_content(&stderr_file_path).await?; - let compilation_info = RustCompilationInfo { - exit_code, - stdout, - stderr, + // Read stdout and stderr + let stdout = read_file_content(&stdout_file_path).await?; + let stderr = read_file_content(&stderr_file_path).await?; + RustCompilationInfo { + exit_code, + stdout, + stderr, + } }; - // Compilation is successful if the return exit code is present and zero - if exit_status.success() { + // Compilation is successful if it exited zero (or rustc was skipped). + if engine_is_crucible || compilation_info.exit_code == 0 { // Source file let source_file_path = workspace_dir .join("target") diff --git a/crates/pipeline-manager/src/compiler/sql_compiler.rs b/crates/pipeline-manager/src/compiler/sql_compiler.rs index a4f81fcd094..18aa781fd2e 100644 --- a/crates/pipeline-manager/src/compiler/sql_compiler.rs +++ b/crates/pipeline-manager/src/compiler/sql_compiler.rs @@ -337,7 +337,11 @@ fn determine_sql_compiler_path( runtime_selector: &RuntimeSelector, ) -> PathBuf { match runtime_selector { - RuntimeSelector::Platform(_) => PathBuf::from(&config.sql_compiler_path), + // Crucible consumes the platform SQL compiler's program info; only the + // Rust half of the compilation is skipped. + RuntimeSelector::Platform(_) | RuntimeSelector::Crucible => { + PathBuf::from(&config.sql_compiler_path) + } RuntimeSelector::Sha(sha) => config .working_dir() .join("sql-compilation") diff --git a/crates/pipeline-manager/src/db/types/program.rs b/crates/pipeline-manager/src/db/types/program.rs index 4057794b639..c109bfb1098 100644 --- a/crates/pipeline-manager/src/db/types/program.rs +++ b/crates/pipeline-manager/src/db/types/program.rs @@ -336,6 +336,11 @@ pub enum RuntimeSelector { /// /// The string corresponds to the git SHA of feldera/feldera at the time of the build. Platform(String), + /// HACK (crucible prototype): the crucible executor runs the pipeline, so + /// no pipeline binary is compiled (a marker placeholder is delivered) and + /// the runner launches the crucible binary. Remove once crucible is the + /// standard executor. + Crucible, } impl From for Option { @@ -344,6 +349,7 @@ impl From for Option { RuntimeSelector::Sha(sha) => Some(sha), RuntimeSelector::Version(version) => Some(version), RuntimeSelector::Platform(_platform_sha) => None, + RuntimeSelector::Crucible => Some("crucible".to_string()), } } } @@ -351,7 +357,7 @@ impl From for Option { impl RuntimeSelector { /// Returns a path to sources of the runtime which the compilation should be using. pub fn runtime_sources(&self, config: &CompilerConfig) -> String { - if self.is_platform() { + if self.is_platform() || self.is_crucible() { config.dbsp_override_path.clone() } else { assert!(has_unstable_feature("runtime_version")); @@ -366,12 +372,18 @@ impl RuntimeSelector { matches!(self, RuntimeSelector::Platform(_)) } + /// Is the crucible executor selected (no pipeline binary compiled)? + pub fn is_crucible(&self) -> bool { + matches!(self, RuntimeSelector::Crucible) + } + /// Bytes representation of the runtime selector (for hashing). pub fn as_bytes(&self) -> &[u8] { match self { RuntimeSelector::Sha(sha) => sha.as_bytes(), RuntimeSelector::Version(version) => version.as_bytes(), RuntimeSelector::Platform(platform_sha) => platform_sha.as_bytes(), + RuntimeSelector::Crucible => b"crucible", } } @@ -381,6 +393,9 @@ impl RuntimeSelector { RuntimeSelector::Sha(sha) => sha, RuntimeSelector::Version(version) => version, RuntimeSelector::Platform(platform_sha) => platform_sha, + RuntimeSelector::Crucible => { + unreachable!("crucible selects no runtime commit; no git operation applies") + } } } } @@ -391,6 +406,7 @@ impl Display for RuntimeSelector { RuntimeSelector::Sha(sha) => write!(f, "{sha}"), RuntimeSelector::Version(version) => write!(f, "{version}"), RuntimeSelector::Platform(platform_sha) => write!(f, "{platform_sha}"), + RuntimeSelector::Crucible => write!(f, "crucible"), } } } @@ -415,7 +431,9 @@ impl TryFrom for RuntimeSelector { version_regex.is_match(s) } - if is_valid_git_sha(&value) { + if value == "crucible" { + Ok(RuntimeSelector::Crucible) + } else if is_valid_git_sha(&value) { Ok(RuntimeSelector::Sha(value)) } else if is_version_tag(&value) { Ok(RuntimeSelector::Version(value)) diff --git a/crates/pipeline-manager/src/runner/local_runner.rs b/crates/pipeline-manager/src/runner/local_runner.rs index 9f09a7f6e7d..ece1cc3a22e 100644 --- a/crates/pipeline-manager/src/runner/local_runner.rs +++ b/crates/pipeline-manager/src/runner/local_runner.rs @@ -39,6 +39,23 @@ use tokio_stream::StreamExt; use tracing::{error, info, warn, Level}; use uuid::Uuid; +/// HACK (crucible prototype): whether the fetched program binary is the +/// compiler's crucible marker placeholder rather than a real pipeline binary. +/// Sniffs only the marker-length prefix, so a real binary costs one small read. +fn binary_is_crucible_placeholder(binary_file_path: &Path) -> bool { + use crate::compiler::rust_compiler::CRUCIBLE_PLACEHOLDER_MARKER; + use std::io::Read; + let Ok(mut file) = std::fs::File::open(binary_file_path) else { + return false; + }; + let mut prefix = vec![0u8; CRUCIBLE_PLACEHOLDER_MARKER.len()]; + match file.read_exact(&mut prefix) { + Ok(()) => prefix == CRUCIBLE_PLACEHOLDER_MARKER, + // Shorter than the marker: not the placeholder. + Err(_) => false, + } +} + /// How many times to attempt to retrieve the pipeline binary. const BINARY_RETRIEVAL_ATTEMPTS: u64 = 5; @@ -1146,7 +1163,23 @@ impl PipelineExecutor for LocalRunner { // - Current directory: pipeline working directory // - Configuration file: path to config.yaml // - Stdout/stderr are piped to follow logs - let mut command = Command::new(&binary_file_path); + // + // HACK (crucible prototype): a program compiled with + // `runtime_version: "crucible"` delivers a marker placeholder + // instead of a pipeline binary; exec the crucible executor for it + // (CRUCIBLE_BINARY overrides the path, default `crucible` on + // PATH). It runs with the same flags and config.yaml, so the + // runner contract is unchanged; pipelines without the marker run + // their compiled binary as always. Remove once crucible is the + // standard executor. + let executable = if binary_is_crucible_placeholder(&binary_file_path) { + std::env::var("CRUCIBLE_BINARY") + .map(std::path::PathBuf::from) + .unwrap_or_else(|_| std::path::PathBuf::from("crucible")) + } else { + binary_file_path.clone() + }; + let mut command = Command::new(&executable); command .env( "TOKIO_WORKER_THREADS", From 73231a4789cd9e43d0c18925b0f0cc7df06c2fc5 Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Tue, 23 Jun 2026 21:45:17 -0700 Subject: [PATCH 4/7] feat(slt): wire the expression-oracle harvest into the slt executor DBSPExecutor.maybeHarvest after the per-query circuit is built, so the sqllogictest suite contributes to the expression-oracle corpus when FELDERA_EXPR_ORACLE_DIR is set (gated; a normal slt run is unaffected). Signed-off-by: Gerd Zellweger --- .../main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java b/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java index b4866613f48..ec84dc68125 100644 --- a/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java +++ b/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java @@ -35,6 +35,7 @@ import org.dbsp.sqlCompiler.circuit.operator.DBSPSinkOperator; import org.dbsp.sqlCompiler.compiler.CompilerOptions; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.backend.ExpressionOracleHarvest; import org.dbsp.sqlCompiler.compiler.backend.rust.RustFileWriter; import org.dbsp.sqlCompiler.compiler.frontend.TableContents; import org.dbsp.sqlCompiler.compiler.frontend.TableData; @@ -301,6 +302,7 @@ ProgramAndTester generateTestCase( DBSPCircuit circuit = compiler.getFinalCircuit(false); Utilities.enforce(circuit != null); circuit.setName("circuit" + suffix); + ExpressionOracleHarvest.maybeHarvest(circuit, compiler); DBSPNode.done(); List sinks = Linq.list(circuit.sinkOperators.values()); From f7996d2f7ad26c40c061b75a67bf6e0b4b5e9d4c Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Wed, 24 Jun 2026 20:59:07 -0700 Subject: [PATCH 5/7] feat(expr-oracle): record generating SQL on harvested closures Each harvest record gains a generating_sql array: the view query the closure feeds, found by walking upstream from every sink. On a dedup collision (an identical closure from a different query) the SQL set is merged, so re-harvesting accumulates provenance instead of overwriting it. Lets a corpus case be traced back to the SQL that produced it. Signed-off-by: Gerd Zellweger --- .../backend/ExpressionOracleHarvest.java | 65 ++++++++++++++++++- 1 file changed, 64 insertions(+), 1 deletion(-) diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java index d6000f15b1c..c3b94849e88 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java @@ -1,10 +1,12 @@ package org.dbsp.sqlCompiler.compiler.backend; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import org.dbsp.sqlCompiler.circuit.DBSPCircuit; import org.dbsp.sqlCompiler.circuit.DBSPDeclaration; +import org.dbsp.sqlCompiler.circuit.OutputPort; import org.dbsp.sqlCompiler.circuit.operator.DBSPFilterOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPJoinOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPLeftJoinOperator; @@ -12,6 +14,7 @@ import org.dbsp.sqlCompiler.circuit.operator.DBSPMapOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSinkOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPStreamJoinOperator; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; import org.dbsp.sqlCompiler.compiler.backend.rust.ToRustInnerVisitor; @@ -31,7 +34,12 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.security.MessageDigest; +import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -95,6 +103,7 @@ public static void maybeHarvest(DBSPCircuit circuit, DBSPCompiler compiler) { private static void harvest(DBSPCircuit circuit, DBSPCompiler compiler, Path dir) throws Exception { Files.createDirectories(dir); + Map> provenance = sqlProvenance(circuit); for (DBSPOperator operator : circuit.getAllOperators()) { if (!(operator instanceof DBSPSimpleOperator simple) || !isHarvestable(operator)) { @@ -132,6 +141,7 @@ private static void harvest(DBSPCircuit circuit, DBSPCompiler compiler, Path dir } String hash = sha1(rustClosure); + Path file = dir.resolve(hash + ".json"); ObjectNode record = MAPPER.createObjectNode(); record.put("name", "case_" + hash); record.put("operator", operator.getClass().getSimpleName()); @@ -140,15 +150,68 @@ private static void harvest(DBSPCircuit circuit, DBSPCompiler compiler, Path dir attachStatics(compiler, declarations, record); record.put("rust_closure", rustClosure); record.set("ir", buildIr(compiler, closure, declarations)); + // Provenance: the SQL view(s) this closure feeds, unioned with whatever a prior + // run recorded under the same dedup file, so a case can be traced to its query. + record.set("generating_sql", + mergeGeneratingSql(file, provenance.getOrDefault(operator, Set.of()))); // Filename is the dedup key: identical closures from different tests and // JVMs converge on one file rather than racing an append. - MAPPER.writeValue(dir.resolve(hash + ".json").toFile(), record); + MAPPER.writeValue(file.toFile(), record); emitted.incrementAndGet(); } writeStats(dir); } + /** + * Map each operator to the SQL of the view(s) it feeds, by walking upstream from every + * sink. One closure can serve several views, so the value is a set of view queries. + */ + private static Map> sqlProvenance(DBSPCircuit circuit) { + Map> provenance = new HashMap<>(); + for (DBSPSinkOperator sink : circuit.sinkOperators.values()) { + String sql = sink.query == null || sink.query.isBlank() + ? sink.viewName.toString() + : sink.query; + Deque work = new ArrayDeque<>(); + Set seen = new HashSet<>(); + work.add(sink); + seen.add(sink); + while (!work.isEmpty()) { + DBSPOperator operator = work.removeFirst(); + provenance.computeIfAbsent(operator, key -> new LinkedHashSet<>()).add(sql); + for (OutputPort input : operator.inputs) { + if (seen.add(input.operator)) { + work.add(input.operator); + } + } + } + } + return provenance; + } + + /** + * Union this closure's view SQL with whatever a prior run already recorded under the same + * dedup file, so re-harvesting accumulates provenance rather than overwriting it. + */ + private static ArrayNode mergeGeneratingSql(Path file, Set fresh) { + LinkedHashSet all = new LinkedHashSet<>(); + if (Files.exists(file)) { + try { + JsonNode prior = MAPPER.readTree(file.toFile()).get("generating_sql"); + if (prior != null && prior.isArray()) { + prior.forEach(node -> all.add(node.asText())); + } + } catch (Exception ignored) { + // A malformed prior file is replaced, not merged. + } + } + all.addAll(fresh); + ArrayNode array = MAPPER.createArrayNode(); + all.forEach(array::add); + return array; + } + /** * Coverage so far, rewritten after every circuit. `candidates` counts the * stateless scalar closures of a harvestable shape; `emitted` counts those that From afd97d9229479ff012654ca09627aff6958ff45e Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Fri, 26 Jun 2026 09:27:02 -0700 Subject: [PATCH 6/7] feat(expr-oracle): harvest array closures and python suite SQL isSupportedLeaf admits Array when the element is itself a supported leaf, so array-of-scalar (and nested-array) closures harvest instead of being skipped. PythonCorpusHarvest compiles the SQL extracted from feldera's python test suites (driven by PYTHON_CORPUS_SQL_DIR) so the oracle harvest hook records their closures; it is skipped when that variable is unset, so a normal mvn test is unaffected. Signed-off-by: Gerd Zellweger --- .../backend/ExpressionOracleHarvest.java | 7 +++ .../compiler/sql/PythonCorpusHarvest.java | 52 +++++++++++++++++++ 2 files changed, 59 insertions(+) create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/PythonCorpusHarvest.java diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java index c3b94849e88..e4807a6f662 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ExpressionOracleHarvest.java @@ -316,6 +316,13 @@ private static ObjectNode tupleParam(DBSPCompiler compiler, DBSPTypeTuple tuple, private static boolean isSupportedLeaf(String rustType) { String base = stripOption(rustType); + // An ARRAY column is `Array`; allow it when the element is itself a supported + // leaf (so the runtime has a Sample/encoding for it). This admits arrays of scalars and + // nested arrays, but not arrays of rows (a `Tup<...>` element has no sampler). + if (base.startsWith("Array<") && base.endsWith(">")) { + String element = base.substring("Array<".length(), base.length() - 1).trim(); + return isSupportedLeaf(element); + } return SUPPORTED_LEAF.contains(base) || base.startsWith("SqlDecimal"); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/PythonCorpusHarvest.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/PythonCorpusHarvest.java new file mode 100644 index 00000000000..ee9241cf442 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/PythonCorpusHarvest.java @@ -0,0 +1,52 @@ +package org.dbsp.sqlCompiler.compiler.sql; + +import org.dbsp.sqlCompiler.compiler.sql.tools.BaseSQLTests; +import org.junit.Assume; +import org.junit.Test; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Stream; + +/** + * Compiles the SQL extracted from feldera's python test suites so the expression-oracle harvest + * hook records their closures. Each {@code .sql} file is one suite's full program (tables plus + * views); {@link BaseSQLTests#getCC} compiles it and {@code getCircuit} fires the harvest. + * + *

Driven by {@code PYTHON_CORPUS_SQL_DIR} (a directory of {@code .sql} files); the test is + * skipped when that is unset, so a normal {@code mvn test} is unaffected. The harvest itself is + * gated on {@code FELDERA_EXPR_ORACLE_DIR}. A file that fails to compile (a suite that expects an + * error, or one using a construct the default compiler options reject) is skipped; the rest still + * harvest. + */ +public class PythonCorpusHarvest extends BaseSQLTests { + @Test + public void harvestPythonCorpus() throws IOException { + String dir = System.getenv("PYTHON_CORPUS_SQL_DIR"); + Assume.assumeTrue("PYTHON_CORPUS_SQL_DIR unset; skipping", dir != null && !dir.isEmpty()); + + List sqlFiles = new ArrayList<>(); + try (Stream entries = Files.list(Path.of(dir))) { + entries.filter(p -> p.toString().endsWith(".sql")).sorted().forEach(sqlFiles::add); + } + + int compiled = 0; + int skipped = 0; + for (Path sqlFile : sqlFiles) { + String sql = Files.readString(sqlFile); + try { + this.getCC(sql); + compiled++; + } catch (Throwable failure) { + skipped++; + System.err.println("PythonCorpusHarvest: skipped " + sqlFile.getFileName() + + ": " + failure.getMessage()); + } + } + System.err.println("PythonCorpusHarvest: compiled " + compiled + " / " + + sqlFiles.size() + " sql file(s), skipped " + skipped); + } +} From 8fe700972cf6dfa07c5117cbcf8abcb8acca1c39 Mon Sep 17 00:00:00 2001 From: Gerd Zellweger Date: Mon, 29 Jun 2026 11:48:57 -0700 Subject: [PATCH 7/7] feat(slt): parallel parse-only harvesting for the expression oracle Gate three changes on FELDERA_EXPR_ORACLE_DIR so a normal SLT run is unchanged. When harvesting: skip a query whose compile fails instead of aborting the 500-query batch, and in parse-only mode (-n) skip writing the generated Rust and copying the temp Cargo.lock the harness never compiles. Together these let the expression-oracle harvest run the whole sqllogictest suite across many shard JVMs that share a working directory without racing on the discarded temp files. Signed-off-by: Gerd Zellweger --- .../main/java/org/dbsp/sqllogictest/Main.java | 21 ++++++++++----- .../sqllogictest/executors/DBSPExecutor.java | 26 ++++++++++++++++--- 2 files changed, 36 insertions(+), 11 deletions(-) diff --git a/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/Main.java b/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/Main.java index 8fef8217c48..e405af6e31f 100644 --- a/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/Main.java +++ b/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/Main.java @@ -147,13 +147,20 @@ public static void main(String[] argv) throws IOException, ClassNotFoundExceptio System.out.println(Arrays.toString(args)); String wd = System.getProperty("user.dir"); System.out.println("WD: " + wd); - Path source = Path.of(wd, "..", "Cargo.lock"); - File sourceFile = source.toFile().getCanonicalFile(); - Path destination = Path.of(wd, "temp", "Cargo.lock"); - File destinationFile = destination.toFile().getCanonicalFile(); - System.out.println("Copying " + sourceFile.getPath() + " to " + destinationFile.getPath()); - Utilities.enforce(sourceFile.exists()); - Files.copy(sourceFile.toPath(), destinationFile.toPath(), StandardCopyOption.REPLACE_EXISTING); + // The temp Cargo.lock feeds the Rust harness compile. Parse-only harvesting never + // compiles it, and parallel harvest shards share a working directory, so skip the copy + // when harvesting to avoid racing on the shared file. + boolean harvesting = System.getenv("FELDERA_EXPR_ORACLE_DIR") != null + && !System.getenv("FELDERA_EXPR_ORACLE_DIR").isBlank(); + if (!harvesting) { + Path source = Path.of(wd, "..", "Cargo.lock"); + File sourceFile = source.toFile().getCanonicalFile(); + Path destination = Path.of(wd, "temp", "Cargo.lock"); + File destinationFile = destination.toFile().getCanonicalFile(); + System.out.println("Copying " + sourceFile.getPath() + " to " + destinationFile.getPath()); + Utilities.enforce(sourceFile.exists()); + Files.copy(sourceFile.toPath(), destinationFile.toPath(), StandardCopyOption.REPLACE_EXISTING); + } OptionsParser parser = new OptionsParser(true, System.out, System.err); parser.registerOption("-skip", "skipCount", "How many tests to skip (for debugging)", o -> { skip.set(Integer.parseInt(o)); diff --git a/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java b/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java index ec84dc68125..7433503881a 100644 --- a/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java +++ b/sql-to-dbsp-compiler/slt/src/main/java/org/dbsp/sqllogictest/executors/DBSPExecutor.java @@ -134,6 +134,12 @@ public void skip(int toSkip) { this.toSkip = toSkip; } + /** Whether closure harvesting is on (the same env var the harvest hook reads). */ + private static boolean harvesting() { + String dir = System.getenv("FELDERA_EXPR_ORACLE_DIR"); + return dir != null && !dir.isBlank(); + } + public TableData[] getInputSets(DBSPCompiler compiler) throws SQLException { for (SltSqlStatement statement : this.inputPreparation.statements) compiler.submitStatementForCompilation(statement.statement); @@ -191,22 +197,34 @@ boolean runBatch(TestStatistics result, boolean cleanup) { result.addFailure( new TestStatistics.FailedTestDescription(testQuery, "Exception during test", "", ex)); + // When harvesting, one query's compile failure must not abort the whole + // batch: skip it so the rest of the file is still harvested. + if (harvesting()) { + queryNo++; + continue; + } throw ex; // return false; } queryNo++; } - // Write the code to Rust files on the filesystem. - DBSPFunction inputFunction = gen.getInputFunction(); - this.writeCodeToFile(compiler, Linq.list(inputFunction), codeGenerated); + // Parse-only harvesting never compiles the generated Rust, so skip writing it: + // the temp/src files are discarded, and parallel harvest shards sharing a working + // directory would otherwise race on them. + boolean harvestOnly = harvesting() && !this.execute; + if (!harvestOnly) { + // Write the code to Rust files on the filesystem. + DBSPFunction inputFunction = gen.getInputFunction(); + this.writeCodeToFile(compiler, Linq.list(inputFunction), codeGenerated); + } this.startTest(); if (this.execute) { Utilities.compileAndTestRust(Main.getAbsoluteRustDirectory(), true); } this.queriesToRun.clear(); System.out.println(elapsedTime(queryNo)); - if (cleanup) + if (cleanup && !harvestOnly) this.cleanupFilesystem(); if (this.execute) // This is not entirely correct, but I am not parsing the rust output