feat: ducklake time-travel UX (snapshot history + AT VERSION reads) (#9709)

* feat: ducklake time-travel UX (snapshot history + AT VERSION reads)

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

* fix: catalog-qualify ducklake time-travel FROM hints (lake. prefix)

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

* fix: render ducklake snapshot_time (microseconds since epoch) correctly

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

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* docs: match History tab UI (master-detail, full-FROM copy) after merge

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

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

---------

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

backend/windmill-common/src/query_builders.rs

@@ -172,6 +172,19 @@ pub struct SimpleColumn {
 pub struct SelectOptions {
     pub limit: Option<i64>,
     pub offset: Option<i64>,
+    /// 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<i64>,
+}
+
+/// 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<i64>) -> 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<bool>,
     ducklake: Option<String>,
+    /// DuckLake snapshot to time-travel the read to (DuckDB only).
+    version: Option<i64>,
 }
 
 #[derive(Deserialize)]
@@ -200,6 +215,21 @@ struct CountPayload {
     #[serde(rename = "whereClause")]
     where_clause: Option<String>,
     ducklake: Option<String>,
+    /// DuckLake snapshot to time-travel the count to (DuckDB only).
+    version: Option<i64>,
+}
+
+/// `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<String>,
 }
 
 #[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<String, String> {
     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<String, String> {
         &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<String, String> {
+    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<ColumnDef> {
     let pks: Vec<ColumnDef> = 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<i64>,
 ) -> Result<String, String> {
     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 "));

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.<table> 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.<table> 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,

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'}
-								<PartitionStatusGrid path={selection.path} {workspace} />
+								<!-- Key on path so switching ducklake assets resets the panel's
+								     selected snapshot / tab instead of carrying state across. -->
+								{#key selection.path}
+									<DucklakeAssetPanel path={selection.path} {workspace} />
+								{/key}
 							{:else}
 								<div class="p-3 text-xs text-secondary">
 									No inline preview yet for {selection.asset_kind}. Use the producer/consumer arrows

frontend/src/lib/components/assets/AssetGraph/DucklakeAssetPanel.svelte

@@ -0,0 +1,116 @@
+<script lang="ts">
+	// Top pane for a selected ducklake asset: tabs over the views of a versioned
+	// table. "Partitions" is the materialization status grid; "History" is a
+	// master-detail time-travel surface — the snapshot list on the left, a
+	// read-only preview of the table at the selected snapshot on the right.
+	//
+	// The snapshot list is fetched here (not in the list child) so both the list
+	// and the preview share one source of truth: `effectiveVersion` derives to
+	// the user's pick, or the newest snapshot until they pick one. The list is
+	// scoped to the table (catalog-wide snapshots predating the table's creation
+	// can't be read), so a selectable version always exists in the table.
+	import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
+	import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
+	import { Table2, History } from 'lucide-svelte'
+	import { Pane, Splitpanes } from 'svelte-splitpanes'
+	import { resource } from 'runed'
+	import { fetchDucklakeSnapshots } from '$lib/components/dbOps'
+	import { parseDbInputFromAssetSyntax } from '$lib/utils'
+	import PartitionStatusGrid from './PartitionStatusGrid.svelte'
+	import DucklakeSnapshotHistory from './DucklakeSnapshotHistory.svelte'
+	import DucklakeVersionPreview from './DucklakeVersionPreview.svelte'
+
+	interface Props {
+		// The materialized ducklake asset path (`<ducklake>/<table>`).
+		path: string
+		workspace: string
+	}
+	let { path, workspace }: Props = $props()
+
+	let tab = $state<'partitions' | 'history'>('partitions')
+	// The user's explicit snapshot pick (undefined until they click a row).
+	let selectedVersion = $state<number | undefined>(undefined)
+
+	let parsed = $derived(parseDbInputFromAssetSyntax(`ducklake://${path}`))
+	let ducklake = $derived(parsed && 'ducklake' in parsed ? parsed.ducklake : undefined)
+	// Schema-qualified table name (schema defaults to `main`) used to scope the
+	// snapshot list to versions where the table exists.
+	let qualifiedTable = $derived.by(() => {
+		if (!parsed || !('specificTable' in parsed) || !parsed.specificTable) return undefined
+		const schema = 'specificSchema' in parsed ? parsed.specificSchema : undefined
+		return `${schema ?? 'main'}.${parsed.specificTable}`
+	})
+
+	let snapshots = resource(
+		[() => workspace, () => ducklake, () => qualifiedTable],
+		async ([ws, dl, table]) => {
+			if (!ws || !dl) return []
+			return await fetchDucklakeSnapshots({ workspace: ws, ducklake: dl, table })
+		}
+	)
+	// Newest snapshot is the default preview until the user picks one. If the
+	// pick is no longer in the list (snapshots refetched, or it was stale from a
+	// previously-viewed asset), fall back to newest rather than reading a version
+	// that isn't in this table.
+	let effectiveVersion = $derived.by(() => {
+		const list = snapshots.current
+		const picked = list?.some((s) => s.snapshot_id === selectedVersion)
+		return picked ? selectedVersion : list?.[0]?.snapshot_id
+	})
+
+	// Keep the list pane to a roughly fixed width rather than a fixed fraction,
+	// so it stays compact on wide panels instead of sprawling. Falls back to a
+	// usable fraction on narrow ones and before the width is measured.
+	let paneWidth = $state(0)
+	let listSize = $derived(
+		paneWidth > 0 ? Math.max(24, Math.min(46, Math.round((260 / paneWidth) * 100))) : 36
+	)
+</script>
+
+<div class="flex flex-col h-full">
+	{#if !qualifiedTable}
+		<!-- Catalog-level ducklake node (no table segment, e.g. `ducklake://main`):
+		     snapshot history / time-travel are per-table, so only the partition
+		     grid applies here. -->
+		<PartitionStatusGrid {path} {workspace} />
+	{:else}
+		<div class="flex items-center gap-2 px-3 py-2 border-b shrink-0">
+			<ToggleButtonGroup selected={tab} on:selected={(e) => (tab = e.detail)}>
+				{#snippet children({ item })}
+					<ToggleButton size="sm" value="partitions" label="Partitions" icon={Table2} {item} />
+					<ToggleButton size="sm" value="history" label="History" icon={History} {item} />
+				{/snippet}
+			</ToggleButtonGroup>
+		</div>
+
+		<div class="flex-1 min-h-0">
+			{#if tab === 'partitions'}
+				<PartitionStatusGrid {path} {workspace} />
+			{:else}
+			<div class="h-full" bind:clientWidth={paneWidth}>
+				<Splitpanes class="!h-full">
+					<Pane size={listSize} minSize={20}>
+						<DucklakeSnapshotHistory
+							items={snapshots.current ?? []}
+							loading={snapshots.loading}
+							error={snapshots.error?.message}
+							onRefresh={() => snapshots.refetch()}
+							selectedVersion={effectiveVersion}
+							onSelect={(v) => (selectedVersion = v)}
+						/>
+					</Pane>
+					<Pane size={100 - listSize} minSize={30}>
+						<div class="h-full p-3 overflow-auto">
+							<DucklakeVersionPreview
+								assetUri={`ducklake://${path}`}
+								version={effectiveVersion}
+								class="h-full"
+							/>
+						</div>
+					</Pane>
+				</Splitpanes>
+			</div>
+		{/if}
+	</div>
+	{/if}
+</div>

frontend/src/lib/components/assets/AssetGraph/DucklakeSnapshotHistory.svelte

@@ -0,0 +1,79 @@
+<script lang="ts">
+	// Time-travel snapshot list for a ducklake table (presentational): the
+	// catalog's DuckLake snapshots, newest first. Selecting a row drives the
+	// adjacent preview pane via `onSelect`; the selected row is highlighted. The
+	// list data + selection default are owned by the parent; the preview pane
+	// carries the copy-able `FROM lake.<table> AT (VERSION => n)` clause.
+	import { Button } from '$lib/components/common'
+	import { Loader2, RefreshCw } from 'lucide-svelte'
+	import type { DucklakeSnapshot } from '$lib/components/dbOps'
+
+	interface Props {
+		items: DucklakeSnapshot[]
+		loading: boolean
+		error?: string
+		onRefresh?: () => void
+		// Snapshot currently shown in the preview pane (highlighted in the list).
+		selectedVersion?: number
+		// Select a snapshot to preview.
+		onSelect?: (version: number) => void
+	}
+	let { items, loading, error, onRefresh, selectedVersion, onSelect }: Props = $props()
+
+	// DuckLake serializes `snapshot_time` as microseconds-since-epoch (a numeric
+	// string), not an ISO date — `new Date(µs)` would be Invalid Date. Detect the
+	// µs magnitude and convert to ms; fall back to direct parsing otherwise.
+	function fmtSnapshotTime(t: string | number | undefined): string {
+		if (t == undefined || t === '') return '—'
+		const n = typeof t === 'number' ? t : Number(t)
+		const d = Number.isFinite(n) && n > 1e14 ? new Date(n / 1000) : new Date(t)
+		return isNaN(d.getTime()) ? String(t) : d.toLocaleString()
+	}
+</script>
+
+<div class="flex flex-col h-full">
+	<div class="flex items-center justify-between gap-2 px-3 py-2 border-b shrink-0">
+		<span class="text-xs font-semibold text-secondary">Snapshot history</span>
+		<Button
+			variant="subtle"
+			unifiedSize="sm"
+			startIcon={{ icon: RefreshCw }}
+			iconOnly
+			onclick={() => onRefresh?.()}
+			title="Refresh"
+		/>
+	</div>
+
+	<div class="flex-1 min-h-0 overflow-auto p-3">
+		{#if loading && !items.length}
+			<div class="flex items-center gap-2 text-tertiary text-xs">
+				<Loader2 size={14} class="animate-spin" /> Loading snapshots…
+			</div>
+		{:else if error}
+			<p class="text-xs text-red-600">Failed to load: {error}</p>
+		{:else if !items.length}
+			<p class="text-xs text-secondary">
+				No snapshots yet. DuckLake records one on every <span class="font-mono"
+					>// materialize</span
+				> write; each becomes a version you can time-travel to.
+			</p>
+		{:else}
+			<div class="flex flex-col gap-1.5">
+				{#each items as s (s.snapshot_id)}
+					{@const selected = s.snapshot_id === selectedVersion}
+					<button
+						type="button"
+						class="flex items-center justify-between gap-2 rounded border px-2 py-1.5 text-left {selected
+							? 'border-blue-400 bg-blue-50 dark:border-blue-500 dark:bg-blue-950/30'
+							: 'hover:bg-surface-hover'}"
+						onclick={() => onSelect?.(s.snapshot_id)}
+						title="Preview the table at this version"
+					>
+						<span class="text-xs font-mono">v{s.snapshot_id}</span>
+						<span class="text-3xs text-tertiary shrink-0">{fmtSnapshotTime(s.snapshot_time)}</span>
+					</button>
+				{/each}
+			</div>
+		{/if}
+	</div>
+</div>

frontend/src/lib/components/assets/AssetGraph/DucklakeVersionPreview.svelte

@@ -0,0 +1,133 @@
+<script lang="ts">
+	// Read-only grid preview of a ducklake table AT a chosen DuckLake snapshot
+	// ("time-travel"). Mirrors DucklakeResultPreview but, instead of scoping to a
+	// partition, pins every read to a catalog version via `AT (VERSION => n)`
+	// (threaded server-side through the SELECT/COUNT markers). This is the
+	// "query this version" scratchpad — the rendered SQL the user would write by
+	// hand is shown above the grid so the affordance is self-documenting.
+	import DBTable from '$lib/components/DBTable.svelte'
+	import { resource } from 'runed'
+	import { workspaceStore } from '$lib/stores'
+	import {
+		fetchDucklakeColumnsAtVersion,
+		dbTableOpsWithPreviewScripts
+	} from '$lib/components/dbOps'
+	import type { DbInput } from '$lib/components/dbTypes'
+	import { parseDbInputFromAssetSyntax, copyToClipboard } from '$lib/utils'
+	import { AlertTriangle, Loader2, ClipboardCopy } from 'lucide-svelte'
+	import { Button } from '$lib/components/common'
+	import { twMerge } from 'tailwind-merge'
+
+	interface Props {
+		// Full asset URI, e.g. `ducklake://main/orders_daily`.
+		assetUri: string
+		// DuckLake snapshot to pin reads to. Undefined renders nothing (the
+		// parent prompts the user to pick a version from the history first).
+		version?: number
+		class?: string
+	}
+	let { assetUri, version, class: className = '' }: Props = $props()
+
+	let input = $derived<DbInput | undefined>(parseDbInputFromAssetSyntax(assetUri) ?? undefined)
+	let ducklake = $derived(input?.type === 'ducklake' ? input.ducklake : undefined)
+	let table = $derived(
+		input && 'specificTable' in input ? (input.specificTable as string | undefined) : undefined
+	)
+	let schema = $derived(
+		input && 'specificSchema' in input ? (input.specificSchema as string | undefined) : undefined
+	)
+	let tableKey = $derived(schema && table ? `${schema}.${table}` : table)
+
+	// A paste-able consumer-script form of this read — `lake` matches the alias
+	// the duckdb scaffold ATTACHes the ducklake under, so the reference is
+	// catalog-qualified — surfaced so the user learns the time-travel syntax.
+	let exampleSql = $derived(
+		version != undefined && tableKey
+			? `FROM lake.${tableKey} AT (VERSION => ${version})`
+			: undefined
+	)
+
+	// Load the column set *at the pinned version* — a table's schema can differ
+	// across snapshots, so enumerating the current columns against an older
+	// version would reference columns that didn't exist then and break the read.
+	// The result carries the version it was loaded for so the read never uses a
+	// stale column set from the previously-viewed snapshot (the resource keeps
+	// the prior value while re-fetching).
+	let columns = resource(
+		() => [ducklake, tableKey, version] as const,
+		async ([_ducklake, _tableKey, _version]) => {
+			if (!_ducklake || !_tableKey || _version == undefined || !$workspaceStore) return undefined
+			const colDefs = await fetchDucklakeColumnsAtVersion({
+				workspace: $workspaceStore,
+				ducklake: _ducklake,
+				tableKey: _tableKey,
+				version: _version
+			})
+			return { version: _version, colDefs }
+		}
+	)
+	// Only ready once the loaded columns are for the version currently shown.
+	let ready = $derived(
+		columns.current?.version === version && (columns.current?.colDefs.length ?? 0) > 0
+	)
+	let tableColDefs = $derived(ready ? columns.current!.colDefs : undefined)
+
+	let dbTableOps = $derived.by(() => {
+		if (!(input && tableColDefs && tableKey && $workspaceStore && version != undefined))
+			return undefined
+		const ops = dbTableOpsWithPreviewScripts({
+			input,
+			tableKey,
+			colDefs: tableColDefs,
+			workspace: $workspaceStore,
+			version
+		})
+		// Historical reads are immutable: drop every mutation handler so DBTable
+		// renders without edit/delete/insert affordances.
+		const readOnly = { ...ops }
+		delete readOnly.onUpdate
+		delete readOnly.onDelete
+		delete readOnly.onInsert
+		return readOnly
+	})
+</script>
+
+<div class={twMerge('flex flex-col min-h-0 relative', className)}>
+	{#if version == undefined}
+		<div class="flex items-center gap-2 p-3 text-2xs text-tertiary">
+			Select a snapshot from the list to preview the table as of that version.
+		</div>
+	{:else}
+		{#if exampleSql}
+			<div class="flex items-center gap-1 pb-2">
+				<span class="text-2xs font-mono text-tertiary truncate" title={exampleSql}>
+					{exampleSql}
+				</span>
+				<Button
+					variant="subtle"
+					unifiedSize="sm"
+					startIcon={{ icon: ClipboardCopy }}
+					iconOnly
+					onclick={() => copyToClipboard(exampleSql)}
+					title="Copy {exampleSql}"
+				/>
+			</div>
+		{/if}
+		{#if ready && dbTableOps}
+			{#key version}
+				<div class="grow min-h-0">
+					<DBTable {dbTableOps} />
+				</div>
+			{/key}
+		{:else if columns.error}
+			<div class="flex items-center gap-2 p-3 text-2xs text-tertiary">
+				<AlertTriangle size={14} class="text-amber-500" />
+				Couldn't load this table at version {version}.
+			</div>
+		{:else}
+			<div class="flex items-center justify-center p-4 text-tertiary">
+				<Loader2 class="animate-spin" size={18} />
+			</div>
+		{/if}
+	{/if}
+</div>

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('')
 

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<ColumnDef[]> {
+	// 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<DucklakeSnapshot[]> {
+	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<void>
 	onCreate: (params: { values: TableEditorValues; schema?: string }) => Promise<void>

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', () => {

frontend/src/lib/utils.ts

@@ -2120,22 +2120,27 @@ export function pick<T extends object, K extends keyof T>(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
 }