From d131d754e1fc9674abf5de383d2bc93596df9bd1 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 24 Jun 2026 23:11:47 +0200 Subject: [PATCH] feat: ducklake time-travel UX (snapshot history + AT VERSION reads) (#9709) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: ducklake time-travel UX (snapshot history + AT VERSION reads) Co-Authored-By: Claude Opus 4.8 (1M context) * fix: catalog-qualify ducklake time-travel FROM hints (lake. prefix) Co-Authored-By: Claude Opus 4.8 (1M context) * fix: render ducklake snapshot_time (microseconds since epoch) correctly Co-Authored-By: Claude Opus 4.8 (1M context) * refactor: merge ducklake History + Query into one master-detail tab Snapshot list (left) selects the version previewed in the read-only grid (right); newest auto-selected. Copy-clause moved to the preview's SQL line. Co-Authored-By: Claude Opus 4.8 (1M context) * fix: scope ducklake snapshot history to the table's versions Catalog-wide snapshots predate a table's creation; previewing AT a version before the table existed errored ("Table ... does not exist at version N"). The DUCKLAKE_SNAPSHOTS marker now takes the table and lists only snapshots from its first creation onward. Also: narrower snapshot-list pane on large screens (target a fixed width, not a fixed fraction). Co-Authored-By: Claude Opus 4.8 (1M context) * fix: load ducklake preview columns at the pinned version + reset on asset switch Addresses CI review (codex/pi P1, cubic P2): - Historical previews loaded current-schema columns, so an AT(VERSION) read enumerating a column added in a later snapshot failed. Now DESCRIBE-loads the column set at the pinned version; the read is gated on columns matching the current version to avoid a stale-colDefs race on version switch. - selectedVersion no longer sticks across assets: the panel is keyed on path (remounts per asset) and effectiveVersion falls back to newest when the pick isn't in the current list. Co-Authored-By: Claude Opus 4.8 (1M context) * docs: match History tab UI (master-detail, full-FROM copy) after merge Co-Authored-By: Claude Opus 4.8 (1M context) * fix: handle catalog-only ducklake asset paths (no table segment) parseDbInputFromAssetSyntax threw on a catalog-only path like 'ducklake://main' (undefined.split('.')) — a real graph node (e.g. a consumer of the whole catalog). It now returns a table-less input instead of throwing, and DucklakeAssetPanel renders only the partition grid (no per-table history/time-travel) for table-less nodes. Adds parser unit tests. Co-Authored-By: Claude Opus 4.8 (1M context) * fix: escape ducklake catalog name in client-built time-travel DESCRIBE fetchDucklakeColumnsAtVersion interpolated the catalog name into an ATTACH string literal without escaping; double single-quotes (mirrors backend escape_sql_literal) so a quote-containing catalog name can't break out. Also fixed the v1.x docs checklist line to match the shipped full-FROM copy affordance. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- backend/windmill-common/src/query_builders.rs | 178 ++++++++++++++++-- docs/ducklake-materialization.md | 66 +++++-- .../AssetGraph/AssetGraphDetailsPane.svelte | 8 +- .../AssetGraph/DucklakeAssetPanel.svelte | 116 ++++++++++++ .../AssetGraph/DucklakeSnapshotHistory.svelte | 79 ++++++++ .../AssetGraph/DucklakeVersionPreview.svelte | 133 +++++++++++++ .../assets/AssetGraph/pipelineTemplates.ts | 8 + frontend/src/lib/components/dbOps.ts | 97 +++++++++- frontend/src/lib/utils.test.ts | 36 +++- frontend/src/lib/utils.ts | 17 +- 10 files changed, 694 insertions(+), 44 deletions(-) create mode 100644 frontend/src/lib/components/assets/AssetGraph/DucklakeAssetPanel.svelte create mode 100644 frontend/src/lib/components/assets/AssetGraph/DucklakeSnapshotHistory.svelte create mode 100644 frontend/src/lib/components/assets/AssetGraph/DucklakeVersionPreview.svelte diff --git a/backend/windmill-common/src/query_builders.rs b/backend/windmill-common/src/query_builders.rs index 2c168cd433..ce7dbc2533 100644 --- a/backend/windmill-common/src/query_builders.rs +++ b/backend/windmill-common/src/query_builders.rs @@ -172,6 +172,19 @@ pub struct SimpleColumn { pub struct SelectOptions { pub limit: Option, pub offset: Option, + /// DuckLake time-travel: when set (DuckDB only), the read is pinned to this + /// catalog snapshot via `AT (VERSION => n)`. Ignored for other db types. + pub version: Option, +} + +/// DuckLake time-travel suffix appended after a table name in a FROM clause. +/// `n` is a server-controlled `i64` (a snapshot id), so inlining it is +/// injection-safe. Empty string when unpinned (reads the latest snapshot). +fn duckdb_version_suffix(version: Option) -> String { + match version { + Some(v) => format!(" AT (VERSION => {})", v), + None => String::new(), + } } // --------------------------------------------------------------------------- @@ -190,6 +203,8 @@ struct SelectPayload { #[serde(rename = "fixPgIntTypes")] fix_pg_int_types: Option, ducklake: Option, + /// DuckLake snapshot to time-travel the read to (DuckDB only). + version: Option, } #[derive(Deserialize)] @@ -200,6 +215,21 @@ struct CountPayload { #[serde(rename = "whereClause")] where_clause: Option, ducklake: Option, + /// DuckLake snapshot to time-travel the count to (DuckDB only). + version: Option, +} + +/// `WM_INTERNAL_DB_DUCKLAKE_SNAPSHOTS` payload — lists the time-travel history +/// of a ducklake table. DuckLake snapshots are catalog-wide commits, so without +/// a `table` this lists every commit; with one it is scoped to snapshots where +/// that table exists (see `expand_ducklake_snapshots`). +#[derive(Deserialize)] +struct DucklakeSnapshotsPayload { + ducklake: String, + /// Schema-qualified table name (e.g. `main.events_daily`) to scope the + /// history to. Snapshots predating the table's creation are excluded — a + /// time-travel read can't target a version where the table didn't exist. + table: Option, } #[derive(Deserialize)] @@ -304,6 +334,10 @@ pub fn try_expand_internal_db_query( expand_primary_key_constraint(json_str, db_type).map(ExpandedQuery::sql) } "SNOWFLAKE_PRIMARY_KEYS" => expand_snowflake_primary_keys(json_str).map(ExpandedQuery::sql), + // DuckLake time-travel: list a ducklake's snapshot history + "DUCKLAKE_SNAPSHOTS" => { + expand_ducklake_snapshots(json_str, db_type).map(ExpandedQuery::sql) + } _ => Err(format!("Unknown WM_INTERNAL_DB operation: {}", op)), }; @@ -324,7 +358,8 @@ fn expand_select(json_str: &str, db_type: DbType) -> Result { let payload: SelectPayload = serde_json::from_str(json_str).map_err(|e| format!("Invalid SELECT payload: {}", e))?; - let options = SelectOptions { limit: payload.limit, offset: payload.offset }; + let options = + SelectOptions { limit: payload.limit, offset: payload.offset, version: payload.version }; let breaking = payload .fix_pg_int_types .map(|v| BreakingFeatures { fix_pg_int_types: v }); @@ -350,11 +385,47 @@ fn expand_count(json_str: &str, db_type: DbType) -> Result { &payload.table, payload.where_clause.as_deref(), &payload.column_defs, + payload.version, )?; Ok(maybe_wrap_ducklake(query, payload.ducklake.as_deref())) } +/// Expand `DUCKLAKE_SNAPSHOTS` into the catalog's time-travel history. DuckLake +/// snapshots are catalog-wide commits, so `ducklake_snapshots('dl')` (the alias +/// `maybe_wrap_ducklake` attaches) lists every version any `AT (VERSION => n)` +/// read can target, newest first. +fn expand_ducklake_snapshots(json_str: &str, db_type: DbType) -> Result { + if db_type != DbType::Duckdb { + return Err("DUCKLAKE_SNAPSHOTS is only supported for DuckDB".to_string()); + } + let payload: DucklakeSnapshotsPayload = serde_json::from_str(json_str) + .map_err(|e| format!("Invalid DUCKLAKE_SNAPSHOTS payload: {}", e))?; + // `dl` is the alias `wrap_ducklake_query` attaches and `USE`s below. + let query = match &payload.table { + // Scope to snapshots from the table's first creation onward. A DuckLake + // table created at snapshot N can't be read before N (the catalog-wide + // list would otherwise offer impossible versions). The creation snapshot + // is the earliest whose `changes.tables_created` names the table; + // COALESCE to 0 (show all) if it is never found. + Some(table) => { + let table = escape_sql_literal(table); + format!( + "SELECT snapshot_id, snapshot_time FROM ducklake_snapshots('dl') \ + WHERE snapshot_id >= COALESCE((\ + SELECT min(snapshot_id) FROM ducklake_snapshots('dl') \ + WHERE list_contains(changes.tables_created, '{table}')), 0) \ + ORDER BY snapshot_id DESC" + ) + } + None => { + "SELECT snapshot_id, snapshot_time FROM ducklake_snapshots('dl') ORDER BY snapshot_id DESC" + .to_string() + } + }; + Ok(maybe_wrap_ducklake(query, Some(&payload.ducklake))) +} + /// Filter columns to primary keys only; fall back to all columns if none are marked. fn pk_columns_or_all(columns: &[ColumnDef]) -> Vec { let pks: Vec = columns.iter().filter(|c| c.isprimarykey).cloned().collect(); @@ -950,9 +1021,10 @@ pub fn make_select_query( ); query.push_str(&format!( - "SELECT {} FROM {}\n", + "SELECT {} FROM {}{}\n", filtered_columns.join(", "), - quote_table_name(table, db_type) + quote_table_name(table, db_type), + duckdb_version_suffix(options.and_then(|o| o.version)) )); query.push_str(&format!( " WHERE {} {}\n", @@ -977,6 +1049,8 @@ pub fn make_count_query( table: &str, where_clause: Option<&str>, column_defs: &[ColumnDef], + // DuckLake time-travel snapshot (DuckDB only); `None` counts the latest. + version: Option, ) -> Result { let where_prefix = " WHERE "; let and_condition = " AND "; @@ -1118,8 +1192,9 @@ pub fn make_count_query( quicksearch_condition.push_str(" ($quicksearch = '' OR 1 = 1)"); } query.push_str(&format!( - "SELECT COUNT(*) as count FROM {}", - quote_table_name(table, db_type) + "SELECT COUNT(*) as count FROM {}{}", + quote_table_name(table, db_type), + duckdb_version_suffix(version) )); } } @@ -2998,7 +3073,7 @@ mod tests { #[test] fn test_select_snowflake_custom_limit() { let cols = vec![col("id", "int")]; - let opts = SelectOptions { limit: Some(50), offset: Some(10) }; + let opts = SelectOptions { limit: Some(50), offset: Some(10), version: None }; let result = make_select_query( "my_table", &cols, @@ -3072,7 +3147,7 @@ mod tests { #[test] fn test_count_postgresql_basic() { let cols = vec![col("id", "int4"), col("name", "text")]; - let result = make_count_query(DbType::Postgresql, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Postgresql, "my_table", None, &cols, None).unwrap(); assert!(result.contains("-- $1 quicksearch (text)")); assert!(result.contains("SELECT COUNT(*) as count FROM \"my_table\"")); @@ -3090,6 +3165,7 @@ mod tests { "my_table", Some("status = 'active'"), &cols, + None, ) .unwrap(); @@ -3105,7 +3181,7 @@ mod tests { c.ignored = Some(true); c }]; - let result = make_count_query(DbType::Postgresql, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Postgresql, "my_table", None, &cols, None).unwrap(); assert!(result.contains("($1 = '' OR 1 = 1)")); } @@ -3116,7 +3192,7 @@ mod tests { #[test] fn test_count_mysql_basic() { let cols = vec![col("id", "int"), col("name", "varchar")]; - let result = make_count_query(DbType::Mysql, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Mysql, "my_table", None, &cols, None).unwrap(); assert!(result.contains("-- :quicksearch (text)")); assert!(result.contains("SELECT COUNT(*) as count FROM `my_table`")); @@ -3130,7 +3206,7 @@ mod tests { #[test] fn test_count_mssql_basic() { let cols = vec![col("id", "int"), col("name", "nvarchar")]; - let result = make_count_query(DbType::MsSqlServer, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::MsSqlServer, "my_table", None, &cols, None).unwrap(); assert!(result.contains("SELECT COUNT(*) as count FROM [my_table]")); assert!(result.contains("(@p1 = '' OR CONCAT([id], [name]) LIKE '%' + @p1 + '%')")); @@ -3143,7 +3219,7 @@ mod tests { #[test] fn test_count_snowflake_basic() { let cols = vec![col("id", "int"), col("name", "text")]; - let result = make_count_query(DbType::Snowflake, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Snowflake, "my_table", None, &cols, None).unwrap(); // Two quicksearch params for snowflake with visible columns assert!(result.contains("-- ? quicksearch (text)\n-- ? quicksearch (text)")); @@ -3158,7 +3234,7 @@ mod tests { c.ignored = Some(true); c }]; - let result = make_count_query(DbType::Snowflake, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Snowflake, "my_table", None, &cols, None).unwrap(); // One quicksearch param let param_lines: Vec<&str> = result.lines().filter(|l| l.starts_with("-- ?")).collect(); assert_eq!(param_lines.len(), 1); @@ -3172,7 +3248,7 @@ mod tests { #[test] fn test_count_bigquery_basic() { let cols = vec![col("id", "INTEGER"), col("name", "STRING")]; - let result = make_count_query(DbType::Bigquery, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Bigquery, "my_table", None, &cols, None).unwrap(); assert!(result.contains("-- @quicksearch (string)")); assert!(result.contains("SELECT COUNT(*) as count FROM `my_table`")); @@ -3182,7 +3258,7 @@ mod tests { #[test] fn test_count_bigquery_json_type() { let cols = vec![col("id", "INTEGER"), col("data", "JSON")]; - let result = make_count_query(DbType::Bigquery, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Bigquery, "my_table", None, &cols, None).unwrap(); assert!(result.contains("TO_JSON_STRING(`data`)")); } @@ -3193,7 +3269,7 @@ mod tests { #[test] fn test_count_duckdb_basic() { let cols = vec![col("id", "int"), col("name", "text")]; - let result = make_count_query(DbType::Duckdb, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Duckdb, "my_table", None, &cols, None).unwrap(); assert!(result.contains("-- $quicksearch (text)")); assert!(result.contains("SELECT COUNT(*) as count FROM \"my_table\"")); @@ -3202,6 +3278,74 @@ mod tests { ); } + // ----------------------------------------------------------------------- + // DuckLake time-travel (AT VERSION) + snapshot history + // ----------------------------------------------------------------------- + + #[test] + fn test_select_duckdb_time_travel() { + let cols = vec![col("id", "int"), col("name", "text")]; + let opts = SelectOptions { limit: None, offset: None, version: Some(42) }; + let result = + make_select_query("orders", &cols, None, DbType::Duckdb, Some(&opts), None).unwrap(); + // Read is pinned to the catalog snapshot via AT (VERSION => n). + assert!(result.contains("FROM \"orders\" AT (VERSION => 42)\n")); + } + + #[test] + fn test_select_duckdb_no_version_unpinned() { + let cols = vec![col("id", "int")]; + let result = make_select_query("orders", &cols, None, DbType::Duckdb, None, None).unwrap(); + // Without a version the read targets the latest snapshot — no AT clause. + assert!(result.contains("FROM \"orders\"\n")); + assert!(!result.contains("AT (VERSION")); + } + + #[test] + fn test_count_duckdb_time_travel() { + let cols = vec![col("id", "int")]; + let result = make_count_query(DbType::Duckdb, "orders", None, &cols, Some(7)).unwrap(); + assert!(result.contains("FROM \"orders\" AT (VERSION => 7)")); + } + + #[test] + fn test_version_ignored_for_non_duckdb() { + // AT (VERSION) is DuckLake-only; other dialects must never emit it even + // if a version is somehow passed through. + let cols = vec![col("id", "int4")]; + let opts = SelectOptions { limit: None, offset: None, version: Some(5) }; + let result = + make_select_query("orders", &cols, None, DbType::Postgresql, Some(&opts), None) + .unwrap(); + assert!(!result.contains("AT (VERSION")); + } + + #[test] + fn test_expand_ducklake_snapshots() { + let json = r#"{"ducklake": "analytics"}"#; + let result = expand_ducklake_snapshots(json, DbType::Duckdb).unwrap(); + assert!(result.contains("ATTACH 'ducklake://analytics' AS dl;USE dl;")); + assert!(result.contains("ducklake_snapshots('dl')")); + assert!(result.contains("ORDER BY snapshot_id DESC")); + // Unscoped: no per-table existence filter. + assert!(!result.contains("tables_created")); + } + + #[test] + fn test_expand_ducklake_snapshots_scoped_to_table() { + let json = r#"{"ducklake": "analytics", "table": "main.events_daily"}"#; + let result = expand_ducklake_snapshots(json, DbType::Duckdb).unwrap(); + // Scoped to snapshots from the table's first creation onward. + assert!(result.contains("list_contains(changes.tables_created, 'main.events_daily')")); + assert!(result.contains("snapshot_id >= COALESCE")); + } + + #[test] + fn test_expand_ducklake_snapshots_non_duckdb_errors() { + let json = r#"{"ducklake": "analytics"}"#; + assert!(expand_ducklake_snapshots(json, DbType::Postgresql).is_err()); + } + // ----------------------------------------------------------------------- // DELETE - all DB types // ----------------------------------------------------------------------- @@ -3500,7 +3644,7 @@ mod tests { #[test] fn test_count_mssql_no_where() { let cols = vec![col("id", "int")]; - let result = make_count_query(DbType::MsSqlServer, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::MsSqlServer, "my_table", None, &cols, None).unwrap(); // MSSQL uses WHERE directly (no AND replacement) assert!(result.contains("SELECT COUNT(*) as count FROM [my_table] WHERE ")); } @@ -3508,7 +3652,7 @@ mod tests { #[test] fn test_count_mysql_no_where_uses_where_keyword() { let cols = vec![col("id", "int")]; - let result = make_count_query(DbType::Mysql, "my_table", None, &cols).unwrap(); + let result = make_count_query(DbType::Mysql, "my_table", None, &cols, None).unwrap(); // The AND should be replaced with WHERE assert!(result.contains("FROM `my_table` WHERE ")); assert!(!result.contains("FROM `my_table` AND ")); diff --git a/docs/ducklake-materialization.md b/docs/ducklake-materialization.md index 3b5c95c770..377931b94a 100644 --- a/docs/ducklake-materialization.md +++ b/docs/ducklake-materialization.md @@ -171,24 +171,58 @@ This one table drives four things at once: ## Reproducibility — the beyond-dbt part -Because every materialization records the snapshot it produced, a downstream -consumer can read the *exact* upstream snapshot its run saw: +Because every materialization records the snapshot it produced, you can read any +table *as of* a past version — a capability dbt has no native answer for (dbt +models are always "whatever's in the warehouse now"). DuckLake gives us this for +the cost of recording one integer per run, and it covers the three things people +actually reach for: **debugging** ("what did this table look like at the failing +run"), **rollback** (re-materialize a consumer from snapshot N), and ad-hoc +**experimentation** on a historical state. + +### How it's surfaced (shipped): explicit, discoverable time-travel + +The version is exposed as a *user-driven* surface, not hidden plumbing. A +consumer pins a read by writing the DuckLake clause directly: ```sql -FROM dl.orders_daily AT (VERSION => $WM_UPSTREAM_SNAPSHOT) +FROM dl.orders_daily AT (VERSION => 42) ``` -The cascade already threads a `trigger` blob (producer path, partition) to each -subscriber; add the producer's captured `snapshot_id` to it, and a consumer's -read is pinned to the upstream state at dispatch time. That makes the *whole -pipeline* reproducible and time-travelable — something dbt has no native answer -for (dbt models are always "whatever's in the warehouse now"). It also gives -rollback (re-point an asset to snapshot N) and "what did this table look like at -the failing run" debugging, for free off the same captured ids. +The asset node's **History** tab is a master-detail view: the snapshot list (id ++ time) on the left selects the version previewed in a read-only grid on the +right, which surfaces — and copies — the catalog-qualified +`FROM lake. AT (VERSION => n)` clause. Snapshot ids are captured automatically; the user opts +into pinning when they want it, and the clause degrades to "latest" if removed, +so the same script still runs standalone. Mechanically this rides on +time-travel **reads** (`make_select_query` / `make_count_query` emit the `AT` +clause when a `version` is threaded through the `WM_INTERNAL_DB_*` markers) plus +a `DUCKLAKE_SNAPSHOTS` read for the history list — capabilities DuckLake already +has, no new write path. -This is the differentiator worth leaning on. It is not catch-up to dbt; it is a -capability dbt structurally cannot offer, and DuckLake gives it to us at the -cost of recording one integer per run. +### Deferred: automatic snapshot pinning across the cascade + +An earlier sketch had the cascade *automatically* thread each producer's +`snapshot_id` into the `trigger` blob and inject `AT (VERSION => $WM_UPSTREAM_SNAPSHOT)` +into consumer reads, so a whole run is pinned to upstream state at dispatch time +without anyone asking. This is deliberately **not** built, for three reasons: + +- **Not critical.** The only thing it adds over the explicit surface above is + *automatic per-run consistency* — protection against an upstream + re-materializing in the window between dispatch and a consumer reading. That + race only bites high-frequency event-driven cascades (rare today), and the + read is always a whole, ACID snapshot regardless — never corruption, just + "newer than the triggering version". Debugging and rollback are already + covered by the explicit surface. +- **Implicit magic.** Auto-injecting an `AT` clause and stripping it on + standalone runs is invisible behaviour to debug when it misfires; the explicit + clause is inspectable. +- **Multi-upstream ambiguity + EE coupling.** A consumer reading two ducklake + upstreams needs a per-ref snapshot *map* accumulated across the AND-join — and + the join-slot logic is EE. A single `$WM_UPSTREAM_SNAPSHOT` would silently pin + every read to one (the firing) producer's snapshot. + +If a workload ever shows the consistency race in practice, pinning can be layered +on top — the capture and the snapshot surfacing built here are its foundation. It also means **we do not build SCD2 snapshots** (gap #4 in `pipelines-vs-dbt.md`): DuckLake time-travel is a strictly better answer for most of what dbt's @@ -224,8 +258,10 @@ Don't try to give both the full treatment for v1. `materialized_partition` rows. 5. **Surface it** — last-materialized/snapshot/row-count on the asset node; missing-partition set feeds the backfill UI. -6. *v1.x* — snapshot pinning across the cascade (`$WM_UPSTREAM_SNAPSHOT`), - rollback, time-travel read helper. +6. *v1.x* — time-travel UX over the captured snapshots: a per-asset **History** + tab — a master-detail snapshot list + query-at-version preview that copies the + full `FROM lake.
AT (VERSION => n)` clause. Automatic cascade pinning + (`$WM_UPSTREAM_SNAPSHOT`) is deferred — see §"Reproducibility" for why. Steps 1–5 are a thin annotation+template layer plus one metadata table and one extra read per run. They deliver managed/incremental/versioned assets, diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 6073ee392c..7560c12a7a 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -29,7 +29,7 @@ import SummaryPathDisplay from '$lib/components/SummaryPathDisplay.svelte' import S3FilePreview from '$lib/components/S3FilePreview.svelte' import DataTablePreview from './DataTablePreview.svelte' - import PartitionStatusGrid from './PartitionStatusGrid.svelte' + import DucklakeAssetPanel from './DucklakeAssetPanel.svelte' import AssetRunsPanel from './AssetRunsPanel.svelte' import { Pane, Splitpanes } from 'svelte-splitpanes' import { fade } from 'svelte/transition' @@ -984,7 +984,11 @@ refreshKey={previewRefreshKey} /> {:else if selection.asset_kind === 'ducklake'} - + + {#key selection.path} + + {/key} {:else}
No inline preview yet for {selection.asset_kind}. Use the producer/consumer arrows diff --git a/frontend/src/lib/components/assets/AssetGraph/DucklakeAssetPanel.svelte b/frontend/src/lib/components/assets/AssetGraph/DucklakeAssetPanel.svelte new file mode 100644 index 0000000000..1e826b4060 --- /dev/null +++ b/frontend/src/lib/components/assets/AssetGraph/DucklakeAssetPanel.svelte @@ -0,0 +1,116 @@ + + +
+ {#if !qualifiedTable} + + + {:else} +
+ (tab = e.detail)}> + {#snippet children({ item })} + + + {/snippet} + +
+ +
+ {#if tab === 'partitions'} + + {:else} +
+ + + snapshots.refetch()} + selectedVersion={effectiveVersion} + onSelect={(v) => (selectedVersion = v)} + /> + + +
+ +
+ + +
+ {/if} +
+ {/if} +
diff --git a/frontend/src/lib/components/assets/AssetGraph/DucklakeSnapshotHistory.svelte b/frontend/src/lib/components/assets/AssetGraph/DucklakeSnapshotHistory.svelte new file mode 100644 index 0000000000..e95be118ee --- /dev/null +++ b/frontend/src/lib/components/assets/AssetGraph/DucklakeSnapshotHistory.svelte @@ -0,0 +1,79 @@ + + +
+
+ Snapshot history +
+ +
+ {#if loading && !items.length} +
+ Loading snapshots… +
+ {:else if error} +

Failed to load: {error}

+ {:else if !items.length} +

+ No snapshots yet. DuckLake records one on every // materialize write; each becomes a version you can time-travel to. +

+ {:else} +
+ {#each items as s (s.snapshot_id)} + {@const selected = s.snapshot_id === selectedVersion} + + {/each} +
+ {/if} +
+
diff --git a/frontend/src/lib/components/assets/AssetGraph/DucklakeVersionPreview.svelte b/frontend/src/lib/components/assets/AssetGraph/DucklakeVersionPreview.svelte new file mode 100644 index 0000000000..5afb51fe66 --- /dev/null +++ b/frontend/src/lib/components/assets/AssetGraph/DucklakeVersionPreview.svelte @@ -0,0 +1,133 @@ + + +
+ {#if version == undefined} +
+ Select a snapshot from the list to preview the table as of that version. +
+ {:else} + {#if exampleSql} +
+ + {exampleSql} + +
+ {/if} + {#if ready && dbTableOps} + {#key version} +
+ +
+ {/key} + {:else if columns.error} +
+ + Couldn't load this table at version {version}. +
+ {:else} +
+ +
+ {/if} + {/if} +
diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts index 380e4300a3..cbb5476516 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts @@ -546,6 +546,14 @@ function bodyDuckdb(ctx: TemplateContext): string { } if (ducklakeDb) { lines.push(`ATTACH 'ducklake://${ducklakeDb}' AS lake;`) + if (input?.kind === 'ducklake') { + // Discoverability hint: every materialize records a DuckLake snapshot, + // so a consumer can pin its read to a past version. Snapshot ids live + // in the asset's History tab. + lines.push( + `-- time-travel: read a past snapshot with \`FROM ${`lake.${catalogTableRef(input.path)}`} AT (VERSION => 42)\`` + ) + } } if (datatableDb || ducklakeDb) lines.push('') diff --git a/frontend/src/lib/components/dbOps.ts b/frontend/src/lib/components/dbOps.ts index 1358951af5..59d19ee172 100644 --- a/frontend/src/lib/components/dbOps.ts +++ b/frontend/src/lib/components/dbOps.ts @@ -1,5 +1,6 @@ import { getLanguageByResourceType, + ColumnIdentity, type ColumnDef, type TableMetadata } from './apps/components/display/dbtable/utils' @@ -46,7 +47,8 @@ export function dbTableOpsWithPreviewScripts({ tableKey, colDefs, workspace, - whereClause + whereClause, + version }: { input: DbInput tableKey: string @@ -55,6 +57,9 @@ export function dbTableOpsWithPreviewScripts({ // Optional raw SQL predicate AND-ed into the read queries (count + rows). // Caller-trusted — build it with escaped values. whereClause?: string + // DuckLake time-travel: when set, reads are pinned to this catalog snapshot + // via `AT (VERSION => n)` (DuckDB/ducklake only). Read-only by nature. + version?: number }): IDbTableOps { const dbType = getDbType(input) const language = getLanguageByResourceType(dbType) @@ -74,7 +79,8 @@ export function dbTableOpsWithPreviewScripts({ const content = makeMarker('COUNT', { table: tableKey, columnDefs: colDefs, - ...(whereClause ? { whereClause } : {}) + ...(whereClause ? { whereClause } : {}), + ...(version != undefined ? { version } : {}) }) const result = await runScriptAndPollResult({ workspace, @@ -88,7 +94,8 @@ export function dbTableOpsWithPreviewScripts({ table: tableKey, columnDefs: colDefs, fixPgIntTypes: true, - ...(whereClause ? { whereClause } : {}) + ...(whereClause ? { whereClause } : {}), + ...(version != undefined ? { version } : {}) }) let items = (await runScriptAndPollResult({ workspace, @@ -131,6 +138,90 @@ export function dbTableOpsWithPreviewScripts({ } } +export type DucklakeSnapshot = { + snapshot_id: number + // DuckLake returns this as microseconds-since-epoch serialized as a string + // (TIMESTAMP); callers must convert before formatting. + snapshot_time: string | number +} + +/** + * Column metadata of a ducklake table *at a specific snapshot*. The catalog's + * `information_schema` only reflects the current schema, so a time-travel read + * pinned to an older version must enumerate the columns that existed *then* — + * otherwise a column added in a later snapshot would break the `SELECT … AT + * (VERSION => n)`. `DESCRIBE SELECT * FROM … AT (VERSION => n)` gives exactly + * that. Returns minimal `ColumnDef`s (field + datatype) — enough for the + * read-only preview's SELECT/COUNT and grid headers. + */ +export async function fetchDucklakeColumnsAtVersion({ + workspace, + ducklake, + tableKey, + version +}: { + workspace: string + ducklake: string + tableKey: string + version: number +}): Promise { + // Quote each identifier part (schema.table) so a dotted/odd table name can't + // break the statement, and double single-quotes in the catalog name so it + // can't break out of the ATTACH string literal (mirrors the backend's + // `escape_sql_literal`). `version` is a number — injection-safe. + const quoted = tableKey + .split('.') + .map((p) => `"${p.replace(/"/g, '""')}"`) + .join('.') + const ducklakeLit = ducklake.replace(/'/g, "''") + const content = + `ATTACH 'ducklake://${ducklakeLit}' AS __dlv__; USE __dlv__; ` + + `DESCRIBE SELECT * FROM ${quoted} AT (VERSION => ${version});` + const rows = (await runScriptAndPollResult({ + workspace, + requestBody: { args: {}, language: 'duckdb', content } + })) as { column_name: string; column_type: string }[] + if (!Array.isArray(rows)) return [] + return rows.map((r) => ({ + field: r.column_name, + datatype: r.column_type, + defaultvalue: '', + isprimarykey: false, + isidentity: ColumnIdentity.No, + isnullable: 'YES' as const, + isenum: false + })) +} + +/** + * List a ducklake table's time-travel history, newest first. DuckLake snapshots + * are catalog-wide commits; passing `table` (schema-qualified, e.g. + * `main.events_daily`) scopes the list to snapshots where the table exists — + * otherwise an `AT (VERSION => n)` read could target a version predating the + * table's creation and error. Runs the `DUCKLAKE_SNAPSHOTS` marker as a duckdb + * preview job (server-side SQL build + ATTACH), so no raw SQL is constructed in + * the client. + */ +export async function fetchDucklakeSnapshots({ + workspace, + ducklake, + table +}: { + workspace: string + ducklake: string + table?: string +}): Promise { + const content = `-- WM_INTERNAL_DB_DUCKLAKE_SNAPSHOTS ${JSON.stringify({ + ducklake, + ...(table ? { table } : {}) + })}` + const rows = await runScriptAndPollResult({ + workspace, + requestBody: { args: {}, language: 'duckdb', content } + }) + return Array.isArray(rows) ? (rows as DucklakeSnapshot[]) : [] +} + export type IDbSchemaOps = { onDelete: (params: { tableKey: string; schema?: string }) => Promise onCreate: (params: { values: TableEditorValues; schema?: string }) => Promise diff --git a/frontend/src/lib/utils.test.ts b/frontend/src/lib/utils.test.ts index 1b2b72a306..dc7e52e9f4 100644 --- a/frontend/src/lib/utils.test.ts +++ b/frontend/src/lib/utils.test.ts @@ -1,5 +1,39 @@ import { describe, it, expect } from 'vitest' -import { cleanValueProperties, getQueryStmtCountHeuristic } from './utils' +import { + cleanValueProperties, + getQueryStmtCountHeuristic, + parseDbInputFromAssetSyntax +} from './utils' + +describe('parseDbInputFromAssetSyntax', () => { + it('parses a table path', () => { + expect(parseDbInputFromAssetSyntax('ducklake://main/orders')).toEqual({ + type: 'ducklake', + ducklake: 'main', + specificTable: 'orders', + specificSchema: undefined + }) + }) + + it('parses a schema-qualified table path', () => { + expect(parseDbInputFromAssetSyntax('ducklake://main/analytics.orders')).toEqual({ + type: 'ducklake', + ducklake: 'main', + specificTable: 'orders', + specificSchema: 'analytics' + }) + }) + + it('handles a catalog-only path without throwing (no table segment)', () => { + // e.g. `// materialize ducklake` → `ducklake://main` — must not throw. + expect(parseDbInputFromAssetSyntax('ducklake://main')).toEqual({ + type: 'ducklake', + ducklake: 'main', + specificTable: undefined, + specificSchema: undefined + }) + }) +}) describe('getQueryStmtCountHeuristic', () => { describe('basic statements', () => { diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index d96cd51b75..fdca33fa22 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -2120,22 +2120,27 @@ export function pick(obj: T, keys: readonly export function parseDbInputFromAssetSyntax(path: string): DbInput | null { const [p1, _p2] = path.split('://') - const [p2, _p3] = _p2.split('/') - const [p3, p4] = _p3.split('.') + const [p2, _p3] = (_p2 ?? '').split('/') + // `_p3` is undefined for a catalog-only path (e.g. `ducklake://main`, no + // table segment) — guard the split so the helper returns a table-less input + // instead of throwing. + const [p3, p4] = (_p3 ?? '').split('.') + const specificTable = p4 || p3 || undefined + const specificSchema = p4 ? p3 : undefined return p1 === 'ducklake' ? { type: 'ducklake', ducklake: p2 || 'main', - specificTable: p4 ?? p3, - specificSchema: p4 ? p3 : undefined + specificTable, + specificSchema } : p1 === 'datatable' ? { type: 'database', resourcePath: `datatable://${p2 || 'main'}`, resourceType: 'postgresql', - specificTable: p4 ?? p3, - specificSchema: p4 ? p3 : undefined + specificTable, + specificSchema } : null }