starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-21 00:02:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(parsers): keep s3 asset path suffix verbatim to preserve storage distinction (#10241)

Browse Source 
* fix(parsers): keep s3 asset path suffix verbatim to preserve storage distinction

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01An2pTqSmqJd2XwnagvX4kM

* package json

* fix(pipelines): preserve named storage in generated TS/Python S3 URIs

The TS/Python templates emitted `s3:///${s3Key(path)}`, stripping the
leading slash and pinning the URI to default storage. For a named-storage
asset path (`secondary/key`) that produced `s3:///secondary/key`, which
resolves to the default storage with key `secondary/key`, dropping the
named-storage dependency and reading/writing the wrong object.

Emit the path verbatim after `s3://` (matching the DuckDB template) so a
named-storage input/output keeps its storage; identical to the previous
output for default-storage paths. Removes the now-unused `s3Key` helper.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* chore(cli): align bun.lock parser versions with frontend

The PR bumped windmill-parser-wasm-asset (1.749.0→1.753.0) and
windmill-parser-wasm-regex (1.692.0→1.764.0) in package.json and the npm
package-lock.json for both cli and frontend, but cli/bun.lock was left
pinned to the old versions. Sync it so the CLI's wasm asset parser (used
by localGraph inference) matches the frontend and deploy-time parser.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
Diego Imbert
2026-07-21 18:25:06 +02:00
committed by GitHub
parent dc5b006e7f
commit 7fb8a2e390
21 changed files with 227 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/parsers/windmill-parser-py-asset/src/lib.rs
+17 -16
View File
@@ -383,7 +383,7 @@ def main():
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "test.csv".to_string(),
path: "/test.csv".to_string(),
access_type: Some(R),
columns: None,
},])
@@ -441,7 +441,7 @@ def main():
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "analytics/x.csv".to_string(),
path: "/analytics/x.csv".to_string(),
access_type: Some(W),
columns: None,
},])
@@ -450,10 +450,11 @@ def main():
#[test]
fn test_py_write_key_matches_duckdb_read_key() {
// Cross-language lineage: this write records `exports/x`, the same path a
// DuckDB `read_csv('s3://exports/x')` resolves to (see
// windmill-parser-sql-asset `test_duckdb_read_key_matches_sdk_write_key`),
// so the producer and consumer connect in the pipeline graph.
// Cross-language lineage: this default-storage write records
// `/exports/x`, the same path a DuckDB `read_csv('s3:///exports/x')`
// resolves to (see windmill-parser-sql-asset
// `test_duckdb_read_key_matches_sdk_write_key`), so the producer and
// consumer connect in the pipeline graph.
let input = r#"
import wmill
from wmill import S3Object
@@ -465,7 +466,7 @@ def main():
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "exports/x".to_string(),
path: "/exports/x".to_string(),
access_type: Some(W),
columns: None,
},])
@@ -508,7 +509,7 @@ def main():
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "dir/in.csv".to_string(),
path: "/dir/in.csv".to_string(),
access_type: Some(R),
columns: None,
},])
@@ -531,14 +532,14 @@ def main():
Ok(vec![
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "mybucket/dir/in.csv".to_string(),
access_type: Some(R),
path: "/out.json".to_string(),
access_type: Some(W),
columns: None,
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "out.json".to_string(),
access_type: Some(W),
path: "mybucket/dir/in.csv".to_string(),
access_type: Some(R),
columns: None,
},
])
@@ -564,25 +565,25 @@ def main():
Ok(vec![
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/enriched.json".to_string(),
path: "/pipelines/km_real/enriched.json".to_string(),
access_type: Some(W),
columns: None,
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/raw_events.json".to_string(),
path: "/pipelines/km_real/raw_events.json".to_string(),
access_type: Some(W),
columns: None,
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/report.json".to_string(),
path: "/pipelines/km_real/report.json".to_string(),
access_type: Some(W),
columns: None,
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/summary.json".to_string(),
path: "/pipelines/km_real/summary.json".to_string(),
access_type: Some(W),
columns: None,
},
backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs
+43 -33
View File
@@ -1261,13 +1261,13 @@ mod tests {
Ok(vec![
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "a.parquet".to_string(),
path: "/a.parquet".to_string(),
access_type: Some(R),
columns: None
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "c.parquet".to_string(),
path: "/c.parquet".to_string(),
access_type: Some(W),
columns: None
},
@@ -1284,25 +1284,35 @@ mod tests {
#[test]
fn test_duckdb_read_key_matches_sdk_write_key() {
// Cross-language lineage: a TS `writeS3File({ s3: "exports/x" })` or
// Python `write_s3_file(S3Object(s3="exports/x"))` records the asset path
// `exports/x` (default storage). A DuckDB reader of the same object must
// resolve to the identical path so the graph connects the producer and
// consumer — both the bare `s3://exports/x` and the triple-slash
// `s3:///exports/x` default-storage form must yield `exports/x`.
for uri in ["s3://exports/x", "s3:///exports/x"] {
let input = format!("SELECT * FROM read_csv('{uri}');");
let assets = parse_assets(&input).expect("parse").assets;
assert_eq!(
assets,
vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "exports/x".to_string(),
access_type: Some(R),
columns: None
}],
"DuckDB read of {uri} must resolve to the SDK write key"
);
}
// Python `write_s3_file(S3Object(s3="exports/x"))` records the asset
// path `/exports/x` (default storage, leading slash). A DuckDB reader
// of the same object uses the triple-slash default-storage URI and
// must resolve to the identical path so the graph connects producer
// and consumer. The bare `s3://exports/x` form names storage
// `exports` instead — a different object, a different path.
let input = "SELECT * FROM read_csv('s3:///exports/x');";
let assets = parse_assets(input).expect("parse").assets;
assert_eq!(
assets,
vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "/exports/x".to_string(),
access_type: Some(R),
columns: None
}],
);
let input = "SELECT * FROM read_csv('s3://exports/x');";
let assets = parse_assets(input).expect("parse").assets;
assert_eq!(
assets,
vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "exports/x".to_string(),
access_type: Some(R),
columns: None
}],
);
}
#[test]
@@ -1318,7 +1328,7 @@ mod tests {
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "out.csv".to_string(),
path: "/out.csv".to_string(),
access_type: Some(W),
columns: None
}])
@@ -1335,7 +1345,7 @@ mod tests {
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "referenced.csv".to_string(),
path: "/referenced.csv".to_string(),
access_type: Some(R),
columns: None
}])
@@ -1355,7 +1365,7 @@ mod tests {
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "data.csv".to_string(),
path: "/data.csv".to_string(),
access_type: Some(RW),
columns: None
}])
@@ -1375,7 +1385,7 @@ mod tests {
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "data.parquet".to_string(),
path: "/data.parquet".to_string(),
access_type: Some(RW),
columns: None
}])
@@ -1393,13 +1403,13 @@ mod tests {
Ok(vec![
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "a.parquet".to_string(),
path: "/a.parquet".to_string(),
access_type: Some(R),
columns: None
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "b.parquet".to_string(),
path: "/b.parquet".to_string(),
access_type: Some(R),
columns: None
}
@@ -1419,7 +1429,7 @@ mod tests {
s.map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "data.parquet".to_string(),
path: "/data.parquet".to_string(),
access_type: Some(RW),
columns: None
}])
@@ -2101,7 +2111,7 @@ mod tests {
assert_eq!(result.len(), 1);
assert_eq!(result[0].kind, AssetKind::S3Object);
assert_eq!(result[0].path, "example_file.parquet");
assert_eq!(result[0].path, "/example_file.parquet");
assert_eq!(result[0].access_type, Some(R));
let columns = result[0].columns.as_ref().expect("Should have columns");
@@ -2132,7 +2142,7 @@ mod tests {
assert_eq!(result.len(), 1);
assert_eq!(result[0].kind, AssetKind::S3Object);
assert_eq!(result[0].path, "example_file.parquet");
assert_eq!(result[0].path, "/example_file.parquet");
assert!(result[0].columns.is_none());
}
@@ -2145,7 +2155,7 @@ mod tests {
assert_eq!(result.len(), 1);
assert_eq!(result[0].kind, AssetKind::S3Object);
assert_eq!(result[0].path, "example_file.parquet");
assert_eq!(result[0].path, "/example_file.parquet");
let columns = result[0].columns.as_ref().expect("Should have columns");
assert_eq!(columns.get("col1"), Some(&R));
@@ -2164,7 +2174,7 @@ mod tests {
assert_eq!(result.len(), 2);
assert!(result.iter().any(|a| {
a.path == "file1.parquet"
a.path == "/file1.parquet"
&& a.columns.as_ref().map_or(false, |c| c.contains_key("col1"))
}));
assert!(result.iter().any(|a| {
@@ -2197,7 +2207,7 @@ mod tests {
assert_eq!(result.len(), 1);
assert_eq!(result[0].kind, AssetKind::S3Object);
assert_eq!(result[0].path, "test.parquet");
assert_eq!(result[0].path, "/test.parquet");
assert_eq!(result[0].access_type, Some(R));
let columns = result[0].columns.as_ref().expect("Should have columns");
backend/parsers/windmill-parser-ts-asset/src/lib.rs
+13 -12
View File
@@ -433,7 +433,7 @@ mod tests {
s.map(|r| r.assets).map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "test.csv".to_string(),
path: "/test.csv".to_string(),
access_type: Some(R),
columns: None,
},])
@@ -461,7 +461,7 @@ mod tests {
s.map(|r| r.assets).map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/raw_events.json".to_string(),
path: "/pipelines/km_real/raw_events.json".to_string(),
access_type: Some(W),
columns: None,
},])
@@ -470,10 +470,11 @@ mod tests {
#[test]
fn test_ts_write_key_matches_duckdb_read_key() {
// Cross-language lineage: this write records `exports/x`, the same path a
// DuckDB `read_csv('s3://exports/x')` resolves to (see
// windmill-parser-sql-asset `test_duckdb_read_key_matches_sdk_write_key`),
// so the producer and consumer connect in the pipeline graph.
// Cross-language lineage: this default-storage write records
// `/exports/x`, the same path a DuckDB `read_csv('s3:///exports/x')`
// resolves to (see windmill-parser-sql-asset
// `test_duckdb_read_key_matches_sdk_write_key`), so the producer and
// consumer connect in the pipeline graph.
let input = r#"
import * as wmill from "windmill-client"
export async function main() {
@@ -485,7 +486,7 @@ mod tests {
s.map(|r| r.assets).map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "exports/x".to_string(),
path: "/exports/x".to_string(),
access_type: Some(W),
columns: None,
},])
@@ -570,25 +571,25 @@ mod tests {
Ok(vec![
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/enriched.json".to_string(),
path: "/pipelines/km_real/enriched.json".to_string(),
access_type: Some(W),
columns: None,
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/raw_events.json".to_string(),
path: "/pipelines/km_real/raw_events.json".to_string(),
access_type: Some(W),
columns: None,
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/report.json".to_string(),
path: "/pipelines/km_real/report.json".to_string(),
access_type: Some(W),
columns: None,
},
ParseAssetsResult {
kind: AssetKind::S3Object,
path: "pipelines/km_real/summary.json".to_string(),
path: "/pipelines/km_real/summary.json".to_string(),
access_type: Some(W),
columns: None,
},
@@ -609,7 +610,7 @@ mod tests {
s.map(|r| r.assets).map_err(|e| e.to_string()),
Ok(vec![ParseAssetsResult {
kind: AssetKind::S3Object,
path: "out.json".to_string(),
path: "/out.json".to_string(),
access_type: Some(W),
columns: None,
},])
backend/parsers/windmill-parser/src/asset_parser.rs
+31 -73
View File
@@ -722,26 +722,13 @@ pub fn parse_asset_syntax(s: &str, enable_default_syntax: bool) -> Option<(Asset
}
for (prefix, kind) in ASSET_KINDS.iter() {
if s.starts_with(prefix) {
let path = &s[prefix.len()..];
// Canonicalize S3 keys to a single asset identity. The SDK object
// form (`{ s3: "key" }` / `S3Object(s3="key")`, default storage)
// resolves to `s3:///key`, whose path is `/key`, while DuckDB
// `s3://key` and `// on s3://key` yield the bare `key`. Strip every
// leading slash so the triple-slash default-storage form and the
// `s3://storage/key` form share one path — otherwise a TS/Python
// writer and a DuckDB reader of the same object become disconnected
// nodes in the pipeline graph. Stripping ALL leading slashes (not
// just one) keeps the identity stable through URI reconstruction:
// `trigger_spec_to_row` rebuilds `s3://<path>`, so a canonical path
// must never itself start with `/` or the rebuilt ref would parse
// back to a different key. Only leading slashes are touched, so
// Hive-partition keys (`s3://b/y=2024/f.parquet`) are untouched.
let path = if matches!(kind, AssetKind::S3Object) {
path.trim_start_matches('/')
} else {
path
};
return Some((*kind, path));
// The suffix is kept verbatim. For S3 the path encodes the storage:
// `s3://<storage>/<key>`, with an EMPTY storage segment for the
// workspace default — so `s3:///key` yields `/key` (leading slash
// significant, default storage) while `s3://secondary/key` yields
// `secondary/key`. Stripping leading slashes here would conflate a
// default-storage object with a named-storage one.
return Some((*kind, &s[prefix.len()..]));
}
}
None
@@ -1560,52 +1547,43 @@ mod pipeline_annotation_tests {
use super::*;
#[test]
fn s3_key_normalization_unifies_uri_forms() {
// A TS/Python SDK write of `{ s3: "exports/x" }` (default storage)
// resolves to the URI `s3:///exports/x`, while a DuckDB read of
// `s3://exports/x` and the `// on s3://exports/x` trigger form yield the
// bare `exports/x`. All three must canonicalize to one asset key so
// the writer and reader connect in the pipeline graph.
let sdk_write = parse_asset_syntax("s3:///exports/x", false);
let duckdb_read = parse_asset_syntax("s3://exports/x", false);
assert_eq!(sdk_write, Some((AssetKind::S3Object, "exports/x")));
assert_eq!(duckdb_read, Some((AssetKind::S3Object, "exports/x")));
assert_eq!(sdk_write, duckdb_read);
fn s3_path_keeps_storage_distinction() {
// An S3 asset path is `<storage>/<key>` with an empty storage segment
// for the workspace default. The default-storage form `s3:///key`
// yields `/key` (leading slash significant); the named-storage form
// `s3://secondary/key` yields `secondary/key`. The two name DIFFERENT
// objects and must never collapse to one identity.
assert_eq!(
parse_asset_syntax("s3:///exports/x", false),
Some((AssetKind::S3Object, "/exports/x"))
);
assert_eq!(
parse_asset_syntax("s3://exports/x", false),
Some((AssetKind::S3Object, "exports/x"))
);
assert_ne!(
parse_asset_syntax("s3:///exports/x", false),
parse_asset_syntax("s3://exports/x", false)
);
// The `// on` trigger annotation goes through the same function.
assert_eq!(
parse_asset_syntax("s3:///exports/x", true),
parse_asset_syntax("s3://exports/x", true)
Some((AssetKind::S3Object, "/exports/x"))
);
// Explicit-storage form is unaffected (no leading slash to strip).
assert_eq!(
parse_asset_syntax("s3://mybucket/exports/x", false),
Some((AssetKind::S3Object, "mybucket/exports/x"))
parse_asset_syntax("s3://secondary_storage/path/to/file.csv", false),
Some((AssetKind::S3Object, "secondary_storage/path/to/file.csv"))
);
// Hive-partition keys and nested paths under default storage are
// preserved verbatim (only leading slashes are stripped).
// Hive-partition keys are preserved verbatim.
assert_eq!(
parse_asset_syntax("s3:///t/year=2024/month=01/f.parquet", false),
Some((AssetKind::S3Object, "t/year=2024/month=01/f.parquet"))
Some((AssetKind::S3Object, "/t/year=2024/month=01/f.parquet"))
);
// Every leading slash is stripped so a canonical S3 path never starts
// with `/`. `S3Object(s3="/x")` resolves to the quad-slash URI
// `s3:////x`; the identity must be the bare `x` (not `/x`) so the ref
// that `trigger_spec_to_row` rebuilds round-trips back to it.
assert_eq!(
parse_asset_syntax("s3:////x", false),
Some((AssetKind::S3Object, "x"))
);
assert_eq!(
parse_asset_syntax("s3://///deep///", false),
Some((AssetKind::S3Object, "deep///"))
);
// Non-S3 kinds keep their leading slash (their paths are workspace-
// relative and the slash is significant).
// Non-S3 kinds also keep their suffix verbatim.
assert_eq!(
parse_asset_syntax("res://f/foo", false),
Some((AssetKind::Resource, "f/foo"))
@@ -1616,26 +1594,6 @@ mod pipeline_annotation_tests {
);
}
#[test]
fn s3_explicit_storage_aliases_default_storage_nested_key() {
// Accepted tradeoff of one canonical key: the explicit-storage form
// `s3://storage/key` and the default-storage nested-key form
// `s3:///storage/key` collapse to the same node `storage/key`, even
// though they name different objects. This is a best-effort lineage
// graph that does not split the first segment as a storage name; the
// collision only happens when a storage config is named to match a
// default-storage prefix. Pinned so the aliasing is intentional, not a
// latent surprise.
assert_eq!(
parse_asset_syntax("s3://mybucket/x", false),
parse_asset_syntax("s3:///mybucket/x", false)
);
assert_eq!(
parse_asset_syntax("s3://mybucket/x", false),
Some((AssetKind::S3Object, "mybucket/x"))
);
}
#[test]
fn bare_pipeline_marker() {
let out = parse_pipeline_annotations("// pipeline\nconsole.log('hi')");
backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json
Vendored
+5 -5
View File
@@ -741,12 +741,12 @@
}
},
{
"name": "s3 triple-slash default-storage trigger canonicalizes to bare key",
"name": "s3 triple-slash default-storage trigger keeps its leading slash",
"code": "// pipeline\n// on s3:///exports/x\nexport function main() {}",
"expected": {
"in_pipeline": true,
"asset_triggers": [
"s3object:exports/x"
"s3object:/exports/x"
],
"native_triggers": [],
"partition": null,
@@ -756,12 +756,12 @@
}
},
{
"name": "s3 quad-slash trigger strips all leading slashes to the bare key",
"code": "// pipeline\n// on s3:////x\nexport function main() {}",
"name": "s3 named-storage trigger keeps the storage segment",
"code": "// pipeline\n// on s3://secondary_storage/exports/x\nexport function main() {}",
"expected": {
"in_pipeline": true,
"asset_triggers": [
"s3object:x"
"s3object:secondary_storage/exports/x"
],
"native_triggers": [],
"partition": null,
backend/windmill-common/src/assets.rs
+6 -9
View File
@@ -567,11 +567,9 @@ mod trigger_ref_roundtrip_tests {
// `trigger_spec_to_row` rebuilds a stored ref as `s3://<path>`, and
// `parse_asset_trigger_ref` parses it back. The two must be inverse for
// every S3 URI form, or a consumer's `// on` trigger lands on a different
// graph node than the producer's inferred write. Because `parse_asset_syntax`
// strips ALL leading slashes, a canonical path never starts with `/`, so the
// naive `prefix + path` rebuild round-trips — including the `S3Object(s3="/x")`
// quad-slash case that previously desynced (path `/x` rebuilt to `s3:///x`,
// which re-parsed to `x`).
// graph node than the producer's inferred write. `parse_asset_syntax`
// keeps the URI suffix verbatim (a default-storage path starts with `/`),
// so the naive `prefix + path` rebuild round-trips for every form.
fn roundtrip(uri: &str) -> String {
let (pkind, path) = parse_asset_syntax(uri, false).expect("parse uri");
assert_eq!(pkind, PAssetKind::S3Object);
@@ -590,11 +588,10 @@ mod trigger_ref_roundtrip_tests {
#[test]
fn s3_trigger_ref_roundtrips_for_every_uri_form() {
assert_eq!(roundtrip("s3:///exports/x"), "exports/x"); // SDK default storage
assert_eq!(roundtrip("s3://exports/x"), "exports/x"); // DuckDB / bare
assert_eq!(roundtrip("s3:///exports/x"), "/exports/x"); // SDK default storage
assert_eq!(roundtrip("s3://exports/x"), "exports/x"); // named storage `exports`
assert_eq!(roundtrip("s3://mybucket/exports/x"), "mybucket/exports/x"); // explicit
assert_eq!(roundtrip("s3:////x"), "x"); // S3Object(s3="/x") quad-slash
assert_eq!(roundtrip("s3:///y=2024/f.parquet"), "y=2024/f.parquet"); // Hive
assert_eq!(roundtrip("s3:///y=2024/f.parquet"), "/y=2024/f.parquet"); // Hive
}
}
cli/bun.lock
+4 -4
View File
@@ -19,7 +19,7 @@
"pg-gateway": "0.3.0-beta.4",
"svelte": "^5.45.2",
"tar-stream": "^3.1.7",
"windmill-parser-wasm-asset": "1.749.0",
"windmill-parser-wasm-asset": "1.753.0",
"windmill-parser-wasm-csharp": "1.510.1",
"windmill-parser-wasm-go": "1.761.0",
"windmill-parser-wasm-java": "1.510.1",
@@ -28,7 +28,7 @@
"windmill-parser-wasm-py": "1.693.1",
"windmill-parser-wasm-py-imports": "1.693.1",
"windmill-parser-wasm-r": "1.668.1",
"windmill-parser-wasm-regex": "1.692.0",
"windmill-parser-wasm-regex": "1.764.0",
"windmill-parser-wasm-ruby": "1.526.1",
"windmill-parser-wasm-rust": "1.647.1",
"windmill-parser-wasm-ts": "1.695.0",
@@ -290,7 +290,7 @@
"utility-types": ["utility-types@3.11.0", "", {}, "sha512-6Z7Ma2aVEWisaL6TvBCy7P8rm2LQoPv6dJ7ecIaIixHcwfbJ0x7mWdbcwlIM5IGQxPZSFYeqRCqlOOeKoJYMkw=="],
"windmill-parser-wasm-asset": ["windmill-parser-wasm-asset@1.749.0", "", {}, "sha512-gj8g9sWQ0tXKfXso7xJxR56sS8Loe/RsnFy+0af5R8siZeCag9ikquGbQ8d8kOqIK2U8eCNA0LqsU/xAFDJIOg=="],
"windmill-parser-wasm-asset": ["windmill-parser-wasm-asset@1.753.0", "", {}, "sha512-zpJhjvcU8EWRoOJzas/nRGKjGdQnvzeB9GOxP+Mdmnk8BFk3uehsmHS2Krxyjo36fUrCHqobbnffEqf0g3LIGg=="],
"windmill-parser-wasm-csharp": ["windmill-parser-wasm-csharp@1.510.1", "", {}, "sha512-qm09YmnbeYHLwYn1jUnObVzPhYO9NZKMlIO7nlo7zPJBXqksgG5fK/KCtwGw9rChrnz+DsvM9wP5FhrwRLMtwQ=="],
@@ -308,7 +308,7 @@
"windmill-parser-wasm-r": ["windmill-parser-wasm-r@1.668.1", "", {}, "sha512-5YNeUibxpNBvYrxCgQcz1PxGhTFx2CyEpg2udtIhq7bx0d4gF/KDZVupMeQmAObmrEtTSFGUWNRJ4zXSWNrSpQ=="],
"windmill-parser-wasm-regex": ["windmill-parser-wasm-regex@1.692.0", "", {}, "sha512-BHGTxrinZJ9ef6hFxbKiBqBEr5uqgG/QySOgMA5r1LswO9n/8fyGswr8JcPT2kGaoeoweV6/RQ+RHVaOhosnKw=="],
"windmill-parser-wasm-regex": ["windmill-parser-wasm-regex@1.764.0", "", {}, "sha512-V2eFdKD90gqWikOvjl2fwMpFqiFt/21+4iQMbiNJYl7Lm2UiEcEZ4r9bpgJLG4TLOLqvD6+u4Ju3WaytxN2O2w=="],
"windmill-parser-wasm-ruby": ["windmill-parser-wasm-ruby@1.526.1", "", {}, "sha512-rMBQA8s21wmL2kA5ztRs/ZgVA3ckxe9/NLjxl3iQPL0CX6DlvfaUH0O+AnhpXXDMyBs1Y1SZIhcnbnvsHZ3R8g=="],
cli/package-lock.json
Generated
+4 -4
View File
@@ -20,7 +20,7 @@
"pg-gateway": "0.3.0-beta.4",
"svelte": "^5.45.2",
"tar-stream": "^3.1.7",
"windmill-parser-wasm-asset": "1.749.0",
"windmill-parser-wasm-asset": "1.753.0",
"windmill-parser-wasm-csharp": "1.510.1",
"windmill-parser-wasm-go": "1.761.0",
"windmill-parser-wasm-java": "1.510.1",
@@ -1413,9 +1413,9 @@
}
},
"node_modules/windmill-parser-wasm-asset": {
"version": "1.749.0",
"resolved": "https://registry.npmjs.org/windmill-parser-wasm-asset/-/windmill-parser-wasm-asset-1.749.0.tgz",
"integrity": "sha512-gj8g9sWQ0tXKfXso7xJxR56sS8Loe/RsnFy+0af5R8siZeCag9ikquGbQ8d8kOqIK2U8eCNA0LqsU/xAFDJIOg=="
"version": "1.753.0",
"resolved": "https://registry.npmjs.org/windmill-parser-wasm-asset/-/windmill-parser-wasm-asset-1.753.0.tgz",
"integrity": "sha512-zpJhjvcU8EWRoOJzas/nRGKjGdQnvzeB9GOxP+Mdmnk8BFk3uehsmHS2Krxyjo36fUrCHqobbnffEqf0g3LIGg=="
},
"node_modules/windmill-parser-wasm-csharp": {
"version": "1.510.1",
cli/package.json
+2 -2
View File
@@ -28,7 +28,7 @@
"pg-gateway": "0.3.0-beta.4",
"svelte": "^5.45.2",
"tar-stream": "^3.1.7",
"windmill-parser-wasm-asset": "1.749.0",
"windmill-parser-wasm-asset": "1.753.0",
"windmill-parser-wasm-csharp": "1.510.1",
"windmill-parser-wasm-go": "1.761.0",
"windmill-parser-wasm-java": "1.510.1",
@@ -54,4 +54,4 @@
"@types/ws": "^8.5.0",
"typescript": "^5.7.0"
}
}
}
cli/src/commands/pipeline/boundedCascade.ts
+4 -6
View File
@@ -87,12 +87,10 @@ export function assetUriToNodeId(uri: string): string | undefined {
if (!m) return undefined;
const prefix = m[1].toLowerCase();
const kind = prefix === "s3" ? "s3object" : prefix;
// Mirror Rust `parse_asset_syntax`: strip all leading slashes from S3 keys so a
// `--to s3:///exports/x` token resolves to the canonical graph node
// `s3object:exports/x` (default storage), same as `s3://exports/x`, and a
// canonical key never starts with `/`.
const path = kind === "s3object" ? m[2].replace(/^\/+/, "") : m[2];
return `${kind}:${path}`;
// The suffix is kept verbatim (mirrors Rust `parse_asset_syntax`): an S3
// path encodes the storage, with a leading `/` for the workspace default
// (`s3:///key` → `/key`) vs `s3://secondary/key` → `secondary/key`.
return `${kind}:${m[2]}`;
}
export type LineageDag = {
cli/src/commands/pipeline/localGraph.ts
+8 -15
View File
@@ -300,13 +300,10 @@ function fallbackParse(content: string, language: string): ParseAssetsRaw {
if (uri) {
const prefix = uri[1].toLowerCase();
const kind = prefix === "s3" ? "s3object" : prefix;
// Mirror Rust `parse_asset_syntax`: strip all leading slashes from S3 keys
// so `s3:///key` (default storage) and `s3://key` / DuckDB canonicalize
// to the same node id (and a canonical key never starts with `/`) —
// otherwise a go/bash fallback consumer's `// on s3:///x` would not
// connect to a wasm-inferred `x` producer.
const path = kind === "s3object" ? uri[2].replace(/^\/+/, "") : uri[2];
out.triggers!.push({ kind: "asset", asset_kind: kind, path });
// The suffix is kept verbatim (mirrors Rust `parse_asset_syntax`): an S3
// path encodes the storage, with a leading `/` for the workspace default
// (`s3:///key` → `/key`) vs `s3://secondary/key` → `secondary/key`.
out.triggers!.push({ kind: "asset", asset_kind: kind, path: uri[2] });
} else if (NATIVE_KINDS.has(firstTok) && rest === firstTok) {
// A native marker (`// on data_upload`) must stand alone: the canonical
// parser rejects a marker line with trailing content (`// on data_upload
@@ -393,14 +390,10 @@ export function parseMuteAnnotations(content: string): {
}
for (const [prefix, kind] of MUTE_ASSET_PREFIXES) {
if (arg.startsWith(prefix)) {
// S3 canonicalization as in `parse_asset_syntax`: strip every leading
// slash so `s3:///key` (default storage) mutes the same node as the
// inferred bare `key`.
const p =
kind === "s3object"
? arg.slice(prefix.length).replace(/^\/+/, "")
: arg.slice(prefix.length);
muted.add(`${kind}:${p}`);
// The suffix is kept verbatim, as in `parse_asset_syntax` — a muted
// `s3:///key` (default storage, path `/key`) only matches an inferred
// default-storage read of the same object.
muted.add(`${kind}:${arg.slice(prefix.length)}`);
break;
}
}
cli/test/pipeline_bounded_cascade_unit.test.ts
+7 -9
View File
@@ -256,16 +256,14 @@ test("assetUriToNodeId maps s3 → s3object, others verbatim", () => {
expect(assetUriToNodeId("nope")).toBe(undefined);
});
test("assetUriToNodeId strips leading slashes from S3 keys (canonical node)", () => {
// Mirror of Rust `parse_asset_syntax`: `--to s3:///exports/x` must resolve to
// the same canonical node as the graph's `s3object:exports/x`.
expect(assetUriToNodeId("s3:///exports/x")).toBe("s3object:exports/x");
expect(assetUriToNodeId("s3:///exports/x")).toBe(assetUriToNodeId("s3://exports/x"));
// All leading slashes stripped so a canonical key never starts with `/`
// (the quad-slash `S3Object(s3="/x")` form collapses to `x`).
expect(assetUriToNodeId("s3:////x")).toBe("s3object:x");
test("assetUriToNodeId keeps the S3 storage distinction (verbatim suffix)", () => {
// Mirror of Rust `parse_asset_syntax`: the suffix is kept verbatim, so a
// default-storage `--to s3:///exports/x` resolves to `s3object:/exports/x`
// while `s3://exports/x` names storage `exports` — a different node.
expect(assetUriToNodeId("s3:///exports/x")).toBe("s3object:/exports/x");
expect(assetUriToNodeId("s3://exports/x")).toBe("s3object:exports/x");
// Hive-partition keys and non-S3 kinds are untouched.
expect(assetUriToNodeId("s3:///t/y=2024/f.parquet")).toBe("s3object:t/y=2024/f.parquet");
expect(assetUriToNodeId("s3:///t/y=2024/f.parquet")).toBe("s3object:/t/y=2024/f.parquet");
});
test("resolveToken: short name, full path, and asset URI", () => {
cli/test/pipeline_local_graph_unit.test.ts
+11 -12
View File
@@ -381,18 +381,19 @@ test("derived triggers dedup against explicit `// on` and never self-trigger a m
test("parseMuteAnnotations mirrors the canonical annotation grammar", () => {
// Any comment prefix regardless of language, header-only scan, complete-word
// keyword, s3 leading-slash canonicalization — in lockstep with the Rust
// keyword, verbatim s3 paths — in lockstep with the Rust
// `parse_pipeline_annotations` / frontend parsePipelineAnnotations.ts.
const all3 = parseMuteAnnotations(
`// mute ducklake://main/a\n-- mute datatable://main/b\n# mute s3:///lead/slash\nSELECT 1;\n// mute ducklake://main/body\n`,
);
expect(all3.muteAll).toBe(false);
// all three prefixes accepted; s3 triple-slash canonicalizes to the bare key;
// the line PAST the first non-comment line is ignored (header-only)
// all three prefixes accepted; the s3 triple-slash default-storage path keeps
// its leading slash; the line PAST the first non-comment line is ignored
// (header-only)
expect([...all3.muted].sort()).toEqual([
"datatable:main/b",
"ducklake:main/a",
"s3object:lead/slash",
"s3object:/lead/slash",
]);
// `mute` must be a complete word, and prose args are not asset URIs
@@ -478,11 +479,11 @@ test("go/bash fallback: leading-header `// on` only, options stripped, no body p
);
});
test("go/bash fallback: `// on s3:///key` canonicalizes to the slashless key", async () => {
// Mirror of the Rust/wasm `parse_asset_syntax` S3 strip: a fallback consumer's
// triple-slash default-storage trigger must resolve to the bare key `exports/x`
// — the same identity a wasm-inferred SDK/DuckDB producer uses — or the local
// graph shows a disconnected `/exports/x` node. Explicit storage is untouched.
test("go/bash fallback: `// on s3:///key` keeps its default-storage leading slash", async () => {
// Mirror of the Rust/wasm `parse_asset_syntax`: the suffix is kept verbatim.
// A triple-slash default-storage trigger resolves to `/exports/x` — the same
// identity a wasm-inferred SDK write of `{ s3: "exports/x" }` uses — while the
// bare `s3://exports/x` names storage `exports` (a different object/node).
await withFolder(
{
"triple.go": `// pipeline\n// on s3:///exports/x\npackage inner\nfunc main() {}\n`,
@@ -495,10 +496,8 @@ test("go/bash fallback: `// on s3:///key` canonicalizes to the slashless key", a
graph.triggers.find(
(t) => t.trigger_kind === "asset" && t.runnable_path === p
) as Extract<(typeof graph.triggers)[number], { trigger_kind: "asset" }> | undefined;
// triple-slash and bare both canonicalize to `exports/x` → same node
expect(pathFor("f/mypipe/triple")?.asset_path).toBe("exports/x");
expect(pathFor("f/mypipe/triple")?.asset_path).toBe("/exports/x");
expect(pathFor("f/mypipe/bare")?.asset_path).toBe("exports/x");
// explicit storage keeps its `storage/key` path (no leading slash to strip)
expect(pathFor("f/mypipe/storage")?.asset_path).toBe("mybucket/exports/x");
},
);
docs/pipeline-local-dev.md
+15 -20
View File
@@ -85,21 +85,16 @@ makes the client own the whole cascade so the backend dispatcher never double-fi
python3, and SQL dialects all get wasm inference (SQL dialects route to `parse_assets_sql`; its
comment-header annotation scan is dialect-independent). `go`/`bash` have no wasm asset parser, so
they fall back to a minimal `// pipeline` + `// on` scan (annotation-only). Inferred asset paths
must match to connect nodes, and all S3 URI forms canonicalize to one key: `parse_asset_syntax`
(shared by the native and wasm parsers) strips leading slashes from S3 paths, so the SDK
object forms — TS `writeS3File({s3:"x"})` and python `write_s3_file(S3Object(s3="x"))`, which
resolve to `s3:///x` (empty default storage) — the triple-slash annotation `// on s3:///x`, DuckDB
`read_csv('s3://x')` / `COPY ... TO 's3://x'`, and `// on s3://x` all yield path `x`. A TS/Python
writer and a DuckDB reader of the same object therefore connect regardless of which URI form each
side uses. (Only leading slashes are stripped — so a canonical key never starts with `/`, which
keeps the identity stable through `// on` trigger-ref reconstruction — while Hive-partition keys
like `s3://bucket/y=2024/f.parquet` and the explicit-storage form `s3://storage/key` keep their
`bucket/…` / `storage/key` paths.) Tradeoff of collapsing to one canonical key: the explicit-storage
form `s3://storage/key` and the default-storage nested-key form `s3:///storage/key` now alias to
the same node `storage/key`, even though they name different objects (a bucket `storage` vs. an
object under the `storage/` prefix in default storage). This only collides when a storage config is
named to match a default-storage prefix — unlikely, and acceptable for a best-effort lineage graph
that already doesn't split the first segment as a storage name.
must match **exactly** to connect nodes. An S3 asset path is `<storage>/<key>` with an empty
storage segment for the workspace default: `parse_asset_syntax` (shared by the native and wasm
parsers) keeps the URI suffix verbatim, so the SDK object forms — TS `writeS3File({s3:"x"})` and
python `write_s3_file(S3Object(s3="x"))`, which resolve to `s3:///x` — the triple-slash annotation
`// on s3:///x`, and DuckDB `read_csv('s3:///x')` / `COPY ... TO 's3:///x'` all yield path `/x`
(leading slash significant). The named-storage form `s3://secondary/key` yields `secondary/key` —
a **different** object and a different node, so the storage distinction is never collapsed. In
practice: use the triple-slash form everywhere for default-storage objects and the producer and
consumer connect regardless of language; mixing in the no-slash form (`s3://x`) names a storage
called `x` and will not connect to a default-storage write.
## How to test
@@ -159,12 +154,12 @@ http://localhost:3000/pipeline_dev?workspace=<WS>&wm_token=<TOK>&folder=demo_pip
`startProxyServer` for embedders that need a localhost origin (e.g. Claude Code preview).
5. **`pipeline dev` editing**: the dev page is view+run only (editing stays in the user's editor).
If in-browser editing with file round-trip is wanted, mirror flow-dev's `handleFlowRoundTrip`.
6. **Asset-path normalization**: done — the python parser resolves the
6. **Asset-path normalization**: the python parser resolves the
`S3Object(s3=…, storage=…?)` constructor / dict-literal forms to the same canonical path as the
TS `{s3, storage}` object form, and `parse_asset_syntax` now strips leading slashes from
S3 keys so the SDK-form `s3:///x` (`/x`) and the bare-URI no-slash form (`x`, DuckDB and
`// on s3://x`) canonicalize to one key (see Language coverage). SDK writes and DuckDB reads of
the same object connect regardless of URI convention.
TS `{s3, storage}` object form, so SDK writes and reads connect across ts/python. The
SDK-form path keeps its default-storage leading slash (`/x`), matching the triple-slash URI
forms (`s3:///x` in DuckDB and `// on s3:///x`); the no-slash `s3://x` form names a storage
called `x` and is intentionally a distinct node (see Language coverage).
## Plan reference
frontend/package-lock.json
Generated
+4 -4
View File
@@ -81,7 +81,7 @@
"vscode-languageclient": "~9.0.1",
"vscode-uri": "~3.1.0",
"vscode-ws-jsonrpc": "~3.5.0",
"windmill-parser-wasm-asset": "1.749.0",
"windmill-parser-wasm-asset": "1.753.0",
"windmill-parser-wasm-csharp": "1.510.1",
"windmill-parser-wasm-go": "1.761.0",
"windmill-parser-wasm-java": "1.510.1",
@@ -14283,9 +14283,9 @@
}
},
"node_modules/windmill-parser-wasm-asset": {
"version": "1.749.0",
"resolved": "https://registry.npmjs.org/windmill-parser-wasm-asset/-/windmill-parser-wasm-asset-1.749.0.tgz",
"integrity": "sha512-gj8g9sWQ0tXKfXso7xJxR56sS8Loe/RsnFy+0af5R8siZeCag9ikquGbQ8d8kOqIK2U8eCNA0LqsU/xAFDJIOg=="
"version": "1.753.0",
"resolved": "https://registry.npmjs.org/windmill-parser-wasm-asset/-/windmill-parser-wasm-asset-1.753.0.tgz",
"integrity": "sha512-zpJhjvcU8EWRoOJzas/nRGKjGdQnvzeB9GOxP+Mdmnk8BFk3uehsmHS2Krxyjo36fUrCHqobbnffEqf0g3LIGg=="
},
"node_modules/windmill-parser-wasm-csharp": {
"version": "1.510.1",
frontend/package.json
+2 -2
View File
@@ -156,7 +156,7 @@
"vscode-languageclient": "~9.0.1",
"vscode-uri": "~3.1.0",
"vscode-ws-jsonrpc": "~3.5.0",
"windmill-parser-wasm-asset": "1.749.0",
"windmill-parser-wasm-asset": "1.753.0",
"windmill-parser-wasm-csharp": "1.510.1",
"windmill-parser-wasm-go": "1.761.0",
"windmill-parser-wasm-java": "1.510.1",
@@ -628,4 +628,4 @@
"@rollup/rollup-linux-x64-gnu": "^4.35.0",
"fsevents": "^2.3.3"
}
}
}
frontend/src/lib/components/assets/AssetGraph/boundedCascade.test.ts
+7 -9
View File
@@ -509,15 +509,13 @@ describe('assetUriToNodeId', () => {
expect(assetUriToNodeId('ducklake://lake/t')).toBe('ducklake:lake/t')
expect(assetUriToNodeId('not-a-uri')).toBeUndefined()
})
it('strips leading slashes from S3 keys so s3:/// and s3:// share a node', () => {
// Mirror of Rust `parse_asset_syntax`: `--to s3:///exports/x` must resolve
// to the same canonical node as the graph's `s3object:exports/x`.
expect(assetUriToNodeId('s3:///exports/x')).toBe('s3object:exports/x')
expect(assetUriToNodeId('s3:///exports/x')).toBe(assetUriToNodeId('s3://exports/x'))
// All leading slashes are stripped so a canonical key never starts with
// `/` (the quad-slash `S3Object(s3="/x")` form collapses to `x`).
expect(assetUriToNodeId('s3:////x')).toBe('s3object:x')
it('keeps the S3 storage distinction (verbatim suffix)', () => {
// Mirror of Rust `parse_asset_syntax`: the suffix is kept verbatim, so a
// default-storage `s3:///exports/x` resolves to `s3object:/exports/x`
// while `s3://exports/x` names storage `exports` — a different node.
expect(assetUriToNodeId('s3:///exports/x')).toBe('s3object:/exports/x')
expect(assetUriToNodeId('s3://exports/x')).toBe('s3object:exports/x')
// Hive-partition keys and non-S3 kinds are untouched.
expect(assetUriToNodeId('s3:///t/y=2024/f.parquet')).toBe('s3object:t/y=2024/f.parquet')
expect(assetUriToNodeId('s3:///t/y=2024/f.parquet')).toBe('s3object:/t/y=2024/f.parquet')
})
})
frontend/src/lib/components/assets/AssetGraph/boundedCascade.ts
+4 -5
View File
@@ -38,11 +38,10 @@ export function assetUriToNodeId(uri: string): string | undefined {
// `s3` is the URI prefix for the `s3object` asset kind (mirrors the CLI
// `assetUri` and the canvas). All other kinds use their name verbatim.
const kind = prefix === 's3' ? 's3object' : prefix
// Mirror Rust `parse_asset_syntax`: strip all leading slashes from S3 keys so
// `s3:///key` (default storage) and `s3://key` resolve to the same node id
// and a canonical key never starts with `/`.
const path = kind === 's3object' ? m[2].replace(/^\/+/, '') : m[2]
return `${kind}:${path}`
// The suffix is kept verbatim (mirrors Rust `parse_asset_syntax`): an S3
// path encodes the storage, with a leading `/` for the workspace default
// (`s3:///key` → `/key`) vs `s3://secondary/key` → `secondary/key`.
return `${kind}:${m[2]}`
}
// Native trigger kinds that fan out *per event*: a single event always flows
frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts
+5 -12
View File
@@ -277,18 +277,11 @@ function stripTrailingKvOpts(s: string): string {
function parseAssetSyntax(s: string): PipelineTriggerAsset | undefined {
for (const [prefix, kind] of ASSET_PREFIXES) {
if (s.startsWith(prefix)) {
let path = s.slice(prefix.length)
// Mirror the Rust `parse_asset_syntax` S3 canonicalization: strip all
// leading slashes so the SDK object form (`s3:///key`, default
// storage) and DuckDB / `// on s3://key` share one asset path, and a
// canonical key never starts with `/` (so ref reconstruction
// round-trips). Without this the live graph preview would show
// disconnected `/key` and `key` nodes. S3-only; leading slashes only,
// so Hive-partition keys are untouched.
if (kind === 's3object') {
path = path.replace(/^\/+/, '')
}
return { kind, path }
// The suffix is kept verbatim, mirroring the Rust `parse_asset_syntax`.
// For S3 the path encodes the storage: `s3:///key` yields `/key`
// (default storage, leading slash significant) while
// `s3://secondary/key` yields `secondary/key` — two different objects.
return { kind, path: s.slice(prefix.length) }
}
}
return undefined
frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.test.ts
+10 -11
View File
@@ -9,16 +9,15 @@ import {
// The seeded draft asset (`autoOutputAsset`, stored as `outputAssets` and used
// by resolveGraph for inactive-draft node identity) must match the asset
// identity the deploy-time / wasm parser infers from the generated body. The
// parser canonicalizes any S3 URI by stripping the `s3://` prefix and all
// leading slashes (see backend `parse_asset_syntax`); if the seed carried a
// leading slash while the body wrote `s3:///key`, the preview would render a
// duplicate `/key` node and a phantom post-deploy drift. This pins the two in
// lockstep so that class of drift can't regress.
// parser keeps the suffix after `s3://` verbatim (see backend
// `parse_asset_syntax`), so a default-storage object's path carries a leading
// slash (`s3:///key` → `/key`). If the seed and the body's write URI disagree,
// the preview renders a duplicate node and a phantom post-deploy drift. This
// pins the two in lockstep so that class of drift can't regress.
// Mirror of the parser's S3 canonicalization for a raw `s3://…` URI.
// Mirror of the parser's S3 path extraction for a raw `s3://…` URI.
function canonicalS3Key(uri: string): string {
const rest = uri.replace(/^s3:\/\//, '')
return rest.replace(/^\/+/, '')
return uri.replace(/^s3:\/\//, '')
}
const S3_KINDS: PipelineOutputKind[] = ['s3_parquet', 's3_object']
@@ -32,10 +31,10 @@ describe('pipelineTemplates S3 seed/body parity', () => {
expect(output).toBeDefined()
const asset = output!
// The seed must be a canonical slashless key so it matches the
// identity the parser infers from the generated body.
// The seed must carry the default-storage leading slash so it
// matches the identity the parser infers from the generated body.
expect(asset.kind).toBe('s3object')
expect(asset.path.startsWith('/')).toBe(false)
expect(asset.path.startsWith('/')).toBe(true)
const body = generatePipelineDraft({
language,
frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts
+25 -30
View File
@@ -175,16 +175,16 @@ export function autoOutputAsset(
case 'ducklake':
case 'materialize':
return { kind: 'ducklake', path: `main/${adj}_${pick(TABLE_NOUNS)}_${slug}` }
// s3 outputs use the canonical slashless key. `parse_asset_syntax`
// normalizes `s3:///<key>` (default storage) and `s3://<key>` to the
// bare `<key>`, so the seeded draft asset must be slashless to match the
// deploy-time inferred identity — otherwise the post-deploy drift check
// would flag the output as a phantom `/`-prefixed node. The generated
// bodies still emit the `s3:///<key>` default-storage URI for runtime I/O.
// s3 paths carry the canonical leading slash of a default-storage
// object (`s3:///<key>` parses to path `/<key>`). The deploy-time
// parser stores writes in that form — a slashless seeded path would
// never match it, and the post-deploy drift check would report the
// output as lost (it isn't; the key differs by one '/'). Bodies emit
// the path verbatim after `s3://`, so the slash round-trips.
case 's3_parquet':
return {
kind: 's3object',
path: `pipelines/${folder}/${adj}_${pick(DATASET_NOUNS)}_${slug}.parquet`
path: `/pipelines/${folder}/${adj}_${pick(DATASET_NOUNS)}_${slug}.parquet`
}
case 's3_object': {
// duckdb's natural output for a generic blob is CSV (one COPY TO
@@ -194,7 +194,7 @@ export function autoOutputAsset(
const ext = language === 'duckdb' ? 'csv' : 'json'
return {
kind: 's3object',
path: `pipelines/${folder}/${adj}_${pick(FILE_NOUNS)}_${slug}.${ext}`
path: `/pipelines/${folder}/${adj}_${pick(FILE_NOUNS)}_${slug}.${ext}`
}
}
// A macro library produces no asset — its "output" is the registry
@@ -222,13 +222,6 @@ export function assetUri(asset: { kind: AssetKind; path: string }): string {
return `${ASSET_URI_PREFIX[asset.kind]}${asset.path}`
}
// Bare object key for the SDK's `{ s3: <key> }` / `s3:///<key>` forms. Asset
// paths are already canonical slashless keys; strip stray leading slashes
// defensively so the emitted key never starts with '/'.
function s3Key(path: string): string {
return path.replace(/^\/+/, '')
}
// Splits a datatable asset path (`<db>/<table>` or `<db>/<schema>.<table>`)
// into its constituent parts. The `<schema>.<table>` grammar is owned by
// `parseDbInputFromAssetSyntax` in $lib/utils.ts (which consumes a full
@@ -438,11 +431,12 @@ function bodyTs(ctx: TemplateContext): string {
if (!input) return ''
switch (input.kind) {
case 's3object':
// `s3:///<key>` URI — one spelling shared with the `// on
// s3:///…` annotation form (the object literal `{ s3: <key> }`
// is equivalent).
// `input.path` encodes storage as `<storage>/<key>` (an empty
// storage segment — leading slash — is the workspace default).
// Emit it verbatim after `s3://` so a named-storage input keeps
// its storage; stripping the slash reads the default-storage key.
return [
` const buf = await wmill.loadS3File(${JSON.stringify(`s3:///${s3Key(input.path)}`)})`,
` const buf = await wmill.loadS3File(${JSON.stringify(`s3://${input.path}`)})`,
` const rows = JSON.parse(new TextDecoder().decode(buf))`,
``
].join('\n')
@@ -468,10 +462,10 @@ function bodyTs(ctx: TemplateContext): string {
switch (outputKind) {
case 's3_parquet':
case 's3_object':
// `s3:///<key>` URI — see the loadS3File note above.
// `s3://<storage>/<key>` URI — see the loadS3File note above.
return [
` const payload = new TextEncoder().encode(JSON.stringify(rows))`,
` await wmill.writeS3File(${JSON.stringify(`s3:///${s3Key(output.path)}`)}, payload)`
` await wmill.writeS3File(${JSON.stringify(`s3://${output.path}`)}, payload)`
].join('\n')
case 'datatable': {
const dbName = output.path.split('/')[0] ?? 'main'
@@ -525,11 +519,12 @@ function bodyPython(ctx: TemplateContext): string {
if (!input) return ''
switch (input.kind) {
case 's3object':
// `s3:///<key>` URI — SDK string params must be s3:// URIs
// (bare keys are rejected), and this form matches the
// `# on s3:///…` annotation spelling.
// SDK string params must be s3:// URIs (bare keys are rejected).
// `input.path` encodes storage as `<storage>/<key>` (empty storage
// segment — leading slash — is the workspace default), so emit it
// verbatim after `s3://` to preserve a named-storage input.
return [
` buf = wmill.load_s3_file(${JSON.stringify(`s3:///${s3Key(input.path)}`)})`,
` buf = wmill.load_s3_file(${JSON.stringify(`s3://${input.path}`)})`,
` import json; rows = json.loads(buf.decode("utf-8"))`
].join('\n')
case 'datatable':
@@ -552,10 +547,10 @@ function bodyPython(ctx: TemplateContext): string {
switch (outputKind) {
case 's3_parquet':
case 's3_object':
// `s3:///<key>` URI — see the load_s3_file note above.
// `s3://<storage>/<key>` URI — see the load_s3_file note above.
return [
` import json`,
` wmill.write_s3_file(${JSON.stringify(`s3:///${s3Key(output.path)}`)}, json.dumps(rows).encode("utf-8"))`
` wmill.write_s3_file(${JSON.stringify(`s3://${output.path}`)}, json.dumps(rows).encode("utf-8"))`
].join('\n')
case 'datatable': {
const dbName = output.path.split('/')[0] ?? 'main'
@@ -647,7 +642,7 @@ function bodyDuckdb(ctx: TemplateContext): string {
if (!input) return null
switch (input.kind) {
case 's3object':
return `read_parquet('s3:///${input.path}')`
return `read_parquet('s3://${input.path}')`
case 'datatable':
// `pg` is the attached Postgres catalog (see ATTACH above).
// Use a 2-part `pg.<table>` ref so the asset parser maps it
@@ -670,7 +665,7 @@ function bodyDuckdb(ctx: TemplateContext): string {
`COPY (`,
` SELECT *`,
` FROM ${fromExpr}`,
`) TO 's3:///${output.path}' (FORMAT 'parquet');`
`) TO 's3://${output.path}' (FORMAT 'parquet');`
)
}
break
@@ -681,7 +676,7 @@ function bodyDuckdb(ctx: TemplateContext): string {
`COPY (`,
` SELECT *`,
` FROM ${fromExpr}`,
`) TO 's3:///${output.path}' (FORMAT 'csv', HEADER);`
`) TO 's3://${output.path}' (FORMAT 'csv', HEADER);`
)
}
break