fix(duckdb): auto-declare partition arg for // partitioned scripts (#9878)

* fix(duckdb): auto-declare the partition arg for // partitioned scripts

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat(cli): pipeline run --arg to pass plain run args to cascade scripts

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

backend/parsers/windmill-parser-sql/src/lib.rs

@@ -826,6 +826,29 @@ fn parse_duckdb_file(code: &str) -> anyhow::Result<Option<Vec<Arg>>> {
     }
 
     args.append(&mut parse_sql_sanitized_interpolation(code));
+
+    // A `// partitioned` script receives its resolved partition as a job arg
+    // named `partition` (windmill_common::partition::PARTITION_ARG), and duckdb
+    // binds named parameters only when they appear in the parsed signature —
+    // so auto-declare it (as `-- $partition (text)` would) to make `$partition`
+    // usable without a manual declaration. An explicit declaration wins.
+    // `has_default` keeps the field optional: the platform resolves the value
+    // at run start when it is not passed explicitly.
+    if !args.iter().any(|arg| arg.name == "partition")
+        && windmill_parser::asset_parser::parse_pipeline_annotations(code)
+            .partition
+            .is_some()
+    {
+        args.push(Arg {
+            name: "partition".to_string(),
+            typ: Typ::Str(None),
+            default: None,
+            otyp: Some("text".to_string()),
+            has_default: true,
+            oidx: None,
+            otyp_inferred: false,
+        });
+    }
     Ok(Some(args))
 }
 
@@ -1985,4 +2008,63 @@ SELECT x
 
         Ok(())
     }
+
+    #[test]
+    fn test_parse_duckdb_partitioned_auto_declares_partition() -> anyhow::Result<()> {
+        let code = r#"// partitioned daily
+// materialize ducklake://main/sales_daily
+SELECT $partition AS day, count(*) AS n FROM sales WHERE day = $partition
+"#;
+        let args = parse_duckdb_sig(code)?.args;
+        assert_eq!(
+            args,
+            vec![Arg {
+                otyp: Some("text".to_string()),
+                name: "partition".to_string(),
+                typ: Typ::Str(None),
+                default: None,
+                has_default: true,
+                oidx: None,
+                otyp_inferred: false,
+            }]
+        );
+
+        // `--`-style annotation headers auto-declare too.
+        let dash_code = "-- partitioned hourly\nSELECT $partition AS h\n";
+        assert_eq!(parse_duckdb_sig(dash_code)?.args, args);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_parse_duckdb_partitioned_explicit_declaration_wins() -> anyhow::Result<()> {
+        let code = r#"// partitioned daily
+-- $partition (text)
+-- $limit (int) = 10
+SELECT * FROM sales WHERE day = $partition LIMIT $limit
+"#;
+        let args = parse_duckdb_sig(code)?.args;
+        // No duplicate: the explicit (required) declaration is kept as-is.
+        assert_eq!(args.iter().filter(|a| a.name == "partition").count(), 1);
+        assert_eq!(
+            args[0],
+            Arg {
+                otyp: Some("text".to_string()),
+                name: "partition".to_string(),
+                typ: Typ::Str(None),
+                default: None,
+                has_default: false,
+                oidx: None,
+                otyp_inferred: false,
+            }
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn test_parse_duckdb_unpartitioned_does_not_declare_partition() -> anyhow::Result<()> {
+        let code = "SELECT 1 AS partition_count\n";
+        assert_eq!(parse_duckdb_sig(code)?.args, vec![]);
+        Ok(())
+    }
 }

backend/windmill-worker/src/duckdb_executor.rs

@@ -1534,6 +1534,39 @@ mod tests {
         assert!(!rewritten.contains("-- $file"));
     }
 
+    // A `// partitioned` script referencing `$partition` needs no manual
+    // `-- $partition (text)` declaration: the parser auto-declares the arg, so
+    // the executor binds the injected `partition` job arg instead of failing
+    // with duckdb's "Wrong number of parameters" at prepare time.
+    #[test]
+    fn partitioned_auto_declares_partition_arg() {
+        let script = "// partitioned daily\n\
+                      // materialize ducklake://main/sales_daily\n\
+                      SELECT $partition AS day, count(*) AS n FROM dl.sales WHERE day = $partition";
+
+        let sig = parse_duckdb_sig(script).expect("sig parses").args;
+        let partition_arg = sig
+            .iter()
+            .find(|a| a.name == "partition")
+            .expect("`partition` auto-declared");
+        assert_eq!(partition_arg.otyp.as_deref(), Some("text"));
+
+        // The wrapped query keeps the `$partition` references so the parsed sig
+        // binds them at run time.
+        let (rewritten, _) = build_materialized_query(
+            script,
+            Some("2026-07-02"),
+            &std::collections::HashMap::new(),
+        )
+        .expect("materialize builds")
+        .expect("materialize present");
+        let rewritten = rewritten.expect("managed mode rewrites the query");
+        assert!(
+            rewritten.contains("$partition"),
+            "wrapped query must keep the `$partition` reference, got:\n{rewritten}"
+        );
+    }
+
     // SCD2 managed mode wraps the SELECT into the diff → close-old → open-new
     // shape (unit-covered in the parser's codegen tests); here we pin the
     // executor-level wiring: the natural key flows through and the wrap is

cli/src/commands/pipeline/pipeline.ts

@@ -40,6 +40,7 @@ import { generatePipelineDocs } from "./docs.ts";
 import {
   type UploadBinding,
   devUploadKey,
+  parseArgBinding,
   parseS3Uri,
   parseUploadBinding,
   s3Arg,
@@ -469,6 +470,7 @@ async function run(
     json?: boolean;
     local?: boolean;
     upload?: string[];
+    arg?: string[];
     defaultTs?: "bun" | "deno";
   },
   folder: string,
@@ -524,6 +526,24 @@ async function run(
     await enrichDeployedNonAutorunTriggers(workspace.workspaceId, graph);
   }
 
+  // Resolve a `--upload`/`--arg` script token to its graph node, with an
+  // actionable error when the short name is ambiguous or matches nothing.
+  const resolveScriptTokenOrThrow = (tok: string, flag: string): string => {
+    const id = resolveToken(graph, tok);
+    if (!id || !id.startsWith("script:")) {
+      const matches = graph.runnables.filter(
+        (r) => r.usage_kind === "script" && (r.path.split("/").pop() ?? r.path) === tok,
+      );
+      if (matches.length > 1) {
+        throw new Error(
+          `${flag} '${tok}' matches multiple scripts (${matches.map((r) => r.path).sort().join(", ")}) — use the full path.`,
+        );
+      }
+      throw new Error(`${flag} '${tok}' matched no script in f/${f}.`);
+    }
+    return id;
+  };
+
   // `--upload <script>[:<param>]=<file|s3://key>` binds an object to a
   // data_upload/webhook entry point so it becomes a runnable start (and its
   // downstream is no longer cut) — the arg is injected at execution. Repeatable
@@ -532,23 +552,28 @@ async function run(
   const boundNodeIds = new Set<string>();
   for (const spec of opts.upload ?? []) {
     const binding = parseUploadBinding(spec);
-    const id = resolveToken(graph, binding.scriptTok);
-    if (!id || !id.startsWith("script:")) {
-      const matches = graph.runnables.filter(
-        (r) => r.usage_kind === "script" && (r.path.split("/").pop() ?? r.path) === binding.scriptTok,
-      );
-      if (matches.length > 1) {
-        throw new Error(
-          `--upload '${binding.scriptTok}' matches multiple scripts (${matches.map((r) => r.path).sort().join(", ")}) — use the full path.`,
-        );
-      }
-      throw new Error(`--upload '${binding.scriptTok}' matched no script in f/${f}.`);
-    }
+    const id = resolveScriptTokenOrThrow(binding.scriptTok, "--upload");
     const p = scriptPathOf(id);
     (boundBindingsByPath.get(p) ?? boundBindingsByPath.set(p, []).get(p)!).push(binding);
     boundNodeIds.add(id);
   }
 
+  // `--arg <script>:<param>=<value>` overlays a plain run arg on a script in
+  // the selection. Unlike `--upload` it does not make the script a runnable
+  // start — it only supplies a value if the script runs.
+  const plainArgsByPath = new Map<string, Record<string, unknown>>();
+  for (const spec of opts.arg ?? []) {
+    const b = parseArgBinding(spec);
+    const p = scriptPathOf(resolveScriptTokenOrThrow(b.scriptTok, "--arg"));
+    const merged = plainArgsByPath.get(p) ?? plainArgsByPath.set(p, {}).get(p)!;
+    // Object.hasOwn, not `in`: a param named like a prototype member
+    // (`toString`, `constructor`, …) must not trip the duplicate check.
+    if (Object.hasOwn(merged, b.param)) {
+      throw new Error(`--arg binds ${p}:${b.param} more than once.`);
+    }
+    merged[b.param] = b.value;
+  }
+
   // Resolve the start: explicit --from (must be a valid start) or the folder's
   // sole valid start. `--upload`-bound scripts join the valid starts.
   const starts = new Set([...validStarts(graph), ...boundNodeIds]);
@@ -687,6 +712,9 @@ async function run(
   for (const p of boundBindingsByPath.keys()) {
     if (!orderSet.has(p)) log.warn(`--upload for ${p} is unused — it isn't in the run selection.`);
   }
+  for (const p of plainArgsByPath.keys()) {
+    if (!orderSet.has(p)) log.warn(`--arg for ${p} is unused — it isn't in the run selection.`);
+  }
 
   if (opts.json) {
     // Surface reachable/dropped ends so a machine-readable plan reflects the
@@ -717,7 +745,9 @@ async function run(
           colors.dim(runAll ? ` (whole pipeline)` : ` (from ${shortName(scriptPathOf(start!))})`),
       );
       order.forEach((p, i) =>
-        log.info(`  ${i + 1}. ${p}${boundBindingsByPath.has(p) ? colors.dim(" (← --upload)") : ""}`),
+        log.info(
+          `  ${i + 1}. ${p}${boundBindingsByPath.has(p) ? colors.dim(" (← --upload)") : ""}${plainArgsByPath.has(p) ? colors.dim(" (← --arg)") : ""}`,
+        ),
       );
     }
     return;
@@ -743,9 +773,14 @@ async function run(
   // Execute in topological order, stopping on the first failure.
   for (const nodePath of order) {
     if (!opts.json) log.info(colors.gray(`▶ running ${nodePath}…`));
-    // Spread the internal dispatch guard LAST so an `--upload`-bound arg can't
+    // `--arg` spreads after `--upload` so an explicit value wins for the same
+    // param; the internal dispatch guard spreads LAST so neither binding can
     // re-enable backend dispatch (the CLI owns the whole closure here).
-    const nodeArgs = { ...(uploadArgs.get(nodePath) ?? {}), _wmill_skip_asset_dispatch: true };
+    const nodeArgs = {
+      ...(uploadArgs.get(nodePath) ?? {}),
+      ...(plainArgsByPath.get(nodePath) ?? {}),
+      _wmill_skip_asset_dispatch: true,
+    };
     let id: string;
     if (opts.local) {
       const ls = localScripts!.get(nodePath);
@@ -827,6 +862,11 @@ const command = new Command()
     "Bind an object to a data_upload/webhook entry point so it runs in the cascade, as SCRIPT[:PARAM]=SOURCE (SOURCE is a local file or an s3://key). Local files are uploaded to the workspace store; the S3Object param is inferred when the script has exactly one. Repeatable.",
     { collect: true },
   )
+  .option(
+    "--arg <binding:string>",
+    "Pass a plain run arg to a script in the cascade, as SCRIPT:PARAM=VALUE (VALUE is parsed as JSON when possible, else taken as a string — e.g. daily_report:partition=2026-07-02). Repeatable.",
+    { collect: true },
+  )
   .action(run as any)
   .command(
     "docs",

cli/src/commands/pipeline/pipelineUpload.ts

@@ -1,6 +1,8 @@
-// Helpers for `wmill pipeline run --upload <script>[:<param>]=<file|s3://key>`:
-// bind an object to a `data_upload` / `webhook` entry point so it (and its
-// downstream) runs in the cascade instead of being skipped for want of input.
+// Helpers for `wmill pipeline run` per-script bindings:
+// `--upload <script>[:<param>]=<file|s3://key>` binds an object to a
+// `data_upload` / `webhook` entry point so it (and its downstream) runs in the
+// cascade instead of being skipped for want of input;
+// `--arg <script>:<param>=<value>` passes a plain (non-S3Object) run arg.
 import { basename } from "node:path";
 
 export type UploadBinding = { scriptTok: string; param?: string; source: string };
@@ -28,6 +30,34 @@ export function parseUploadBinding(spec: string): UploadBinding {
   return { scriptTok, param, source };
 }
 
+export type ArgBinding = { scriptTok: string; param: string; value: unknown };
+
+/**
+ * Parse an `--arg` spec: `<script>:<param>=<value>`. Split on the FIRST `=`
+ * (values may contain `=`); the value is JSON when it parses as JSON, else the
+ * raw string — so `stats:limit=10` is a number and
+ * `daily_report:partition=2026-07-02` a string. `:<param>` is required: unlike
+ * `--upload` there is no single-S3Object convention to infer it from.
+ */
+export function parseArgBinding(spec: string): ArgBinding {
+  const eq = spec.indexOf("=");
+  const left = eq < 0 ? "" : spec.slice(0, eq).trim();
+  const raw = eq < 0 ? "" : spec.slice(eq + 1).trim();
+  const colon = left.indexOf(":");
+  const scriptTok = colon < 0 ? "" : left.slice(0, colon).trim();
+  const param = colon < 0 ? "" : left.slice(colon + 1).trim();
+  if (!scriptTok || !param) {
+    throw new Error(`--arg '${spec}' must be <script>:<param>=<json-or-string>`);
+  }
+  let value: unknown;
+  try {
+    value = JSON.parse(raw);
+  } catch {
+    value = raw;
+  }
+  return { scriptTok, param, value };
+}
+
 /** Names of a schema's S3Object properties (`format: resource-s3_object`). */
 export function s3ObjectParams(schema: any): string[] {
   const props = schema?.properties ?? {};

cli/src/guidance/skills.gen.ts

@@ -6925,6 +6925,7 @@ inspect asset-driven pipelines (scripts marked \`// pipeline\`, wired by \`// on
   - \`--json\` - Output the plan as JSON (for piping to jq).
   - \`--local\` - Run the local working-tree scripts via preview (no deploy) instead of the deployed versions; the graph is built from local files.
   - \`--upload <binding:string>\` - Bind an object to a data_upload/webhook entry point so it runs in the cascade, as SCRIPT[:PARAM]=SOURCE (SOURCE is a local file or an s3://key). Local files are uploaded to the workspace store; the S3Object param is inferred when the script has exactly one. Repeatable.
+  - \`--arg <binding:string>\` - Pass a plain run arg to a script in the cascade, as SCRIPT:PARAM=VALUE (VALUE is parsed as JSON when possible, else taken as a string — e.g. daily_report:partition=2026-07-02). Repeatable.
 - \`pipeline docs <folder:string>\` - generate PIPELINE.md (+ AGENTS.md pointer) describing a folder's pipeline graph and datatable schemas, for an editor / agentic loop
   - \`--local\` - Build the graph from local working-tree files instead of the deployed workspace.
 - \`pipeline dev [folder:string]\` - Live-preview a data pipeline from local files: watch an \`f/<folder>\` of \`// pipeline\` scripts, push the working-tree graph to the dev page, and run the cascade via preview (no deploy).

cli/test/pipeline_upload_unit.test.ts

@@ -1,6 +1,7 @@
 import { expect, test } from "bun:test";
 import {
   devUploadKey,
+  parseArgBinding,
   parseS3Uri,
   parseUploadBinding,
   s3Arg,
@@ -42,6 +43,50 @@ test("parseUploadBinding: malformed specs throw", () => {
   expect(() => parseUploadBinding(":file=./x.csv")).toThrow(); // empty script
 });
 
+test("parseArgBinding: JSON values parse, non-JSON falls back to the raw string", () => {
+  // a bare date is not valid JSON → string
+  expect(parseArgBinding("daily_report:partition=2026-07-02")).toEqual({
+    scriptTok: "daily_report",
+    param: "partition",
+    value: "2026-07-02",
+  });
+  // JSON scalars/objects keep their type
+  expect(parseArgBinding("stats:limit=10")).toEqual({
+    scriptTok: "stats",
+    param: "limit",
+    value: 10,
+  });
+  expect(parseArgBinding("stats:enabled=true")).toEqual({
+    scriptTok: "stats",
+    param: "enabled",
+    value: true,
+  });
+  expect(parseArgBinding('f/pd/stats:opts={"a":1}')).toEqual({
+    scriptTok: "f/pd/stats",
+    param: "opts",
+    value: { a: 1 },
+  });
+  // quoting forces a string even for number-looking values
+  expect(parseArgBinding('stats:code="42"')).toEqual({
+    scriptTok: "stats",
+    param: "code",
+    value: "42",
+  });
+  // split on the FIRST `=` — values may contain `=`
+  expect(parseArgBinding("stats:expr=a=b")).toEqual({
+    scriptTok: "stats",
+    param: "expr",
+    value: "a=b",
+  });
+});
+
+test("parseArgBinding: :param is required and malformed specs throw", () => {
+  expect(() => parseArgBinding("stats=10")).toThrow(); // no :param
+  expect(() => parseArgBinding("stats:limit")).toThrow(); // no `=`
+  expect(() => parseArgBinding(":limit=10")).toThrow(); // empty script
+  expect(() => parseArgBinding("stats:=10")).toThrow(); // empty param
+});
+
 test("s3ObjectParams: only resource-s3_object properties, in declaration order", () => {
   const schema = {
     properties: {

system_prompts/auto-generated/cli/cli-commands.md

@@ -427,6 +427,7 @@ inspect asset-driven pipelines (scripts marked `// pipeline`, wired by `// on <s
   - `--json` - Output the plan as JSON (for piping to jq).
   - `--local` - Run the local working-tree scripts via preview (no deploy) instead of the deployed versions; the graph is built from local files.
   - `--upload <binding:string>` - Bind an object to a data_upload/webhook entry point so it runs in the cascade, as SCRIPT[:PARAM]=SOURCE (SOURCE is a local file or an s3://key). Local files are uploaded to the workspace store; the S3Object param is inferred when the script has exactly one. Repeatable.
+  - `--arg <binding:string>` - Pass a plain run arg to a script in the cascade, as SCRIPT:PARAM=VALUE (VALUE is parsed as JSON when possible, else taken as a string — e.g. daily_report:partition=2026-07-02). Repeatable.
 - `pipeline docs <folder:string>` - generate PIPELINE.md (+ AGENTS.md pointer) describing a folder's pipeline graph and datatable schemas, for an editor / agentic loop
   - `--local` - Build the graph from local working-tree files instead of the deployed workspace.
 - `pipeline dev [folder:string]` - Live-preview a data pipeline from local files: watch an `f/<folder>` of `// pipeline` scripts, push the working-tree graph to the dev page, and run the cascade via preview (no deploy).

system_prompts/auto-generated/prompts.ts

@@ -3095,6 +3095,7 @@ inspect asset-driven pipelines (scripts marked \`// pipeline\`, wired by \`// on
   - \`--json\` - Output the plan as JSON (for piping to jq).
   - \`--local\` - Run the local working-tree scripts via preview (no deploy) instead of the deployed versions; the graph is built from local files.
   - \`--upload <binding:string>\` - Bind an object to a data_upload/webhook entry point so it runs in the cascade, as SCRIPT[:PARAM]=SOURCE (SOURCE is a local file or an s3://key). Local files are uploaded to the workspace store; the S3Object param is inferred when the script has exactly one. Repeatable.
+  - \`--arg <binding:string>\` - Pass a plain run arg to a script in the cascade, as SCRIPT:PARAM=VALUE (VALUE is parsed as JSON when possible, else taken as a string — e.g. daily_report:partition=2026-07-02). Repeatable.
 - \`pipeline docs <folder:string>\` - generate PIPELINE.md (+ AGENTS.md pointer) describing a folder's pipeline graph and datatable schemas, for an editor / agentic loop
   - \`--local\` - Build the graph from local working-tree files instead of the deployed workspace.
 - \`pipeline dev [folder:string]\` - Live-preview a data pipeline from local files: watch an \`f/<folder>\` of \`// pipeline\` scripts, push the working-tree graph to the dev page, and run the cascade via preview (no deploy).

system_prompts/auto-generated/skills/cli-commands/SKILL.md

@@ -432,6 +432,7 @@ inspect asset-driven pipelines (scripts marked `// pipeline`, wired by `// on <s
   - `--json` - Output the plan as JSON (for piping to jq).
   - `--local` - Run the local working-tree scripts via preview (no deploy) instead of the deployed versions; the graph is built from local files.
   - `--upload <binding:string>` - Bind an object to a data_upload/webhook entry point so it runs in the cascade, as SCRIPT[:PARAM]=SOURCE (SOURCE is a local file or an s3://key). Local files are uploaded to the workspace store; the S3Object param is inferred when the script has exactly one. Repeatable.
+  - `--arg <binding:string>` - Pass a plain run arg to a script in the cascade, as SCRIPT:PARAM=VALUE (VALUE is parsed as JSON when possible, else taken as a string — e.g. daily_report:partition=2026-07-02). Repeatable.
 - `pipeline docs <folder:string>` - generate PIPELINE.md (+ AGENTS.md pointer) describing a folder's pipeline graph and datatable schemas, for an editor / agentic loop
   - `--local` - Build the graph from local working-tree files instead of the deployed workspace.
 - `pipeline dev [folder:string]` - Live-preview a data pipeline from local files: watch an `f/<folder>` of `// pipeline` scripts, push the working-tree graph to the dev page, and run the cascade via preview (no deploy).