feat(pipelines): self-teaching custom data_test errors + scaffold (#9937)

Custom `// data_test <path>` scripts must be a single SELECT reading the
freshly-materialized target via the internal `_wm_target.<table>` alias —
neither was documented or scaffolded. Make the codegen errors name the exact
violation (multi-statement, non-SELECT, wrong alias, empty) and append a
copyable `SELECT * FROM _wm_target.<table> WHERE <condition>` example. Add a
DuckDB-only 'Data test' pipeline output kind that scaffolds that starter body.

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

frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.dataTest.test.ts

@@ -0,0 +1,56 @@
+import { describe, expect, it } from 'vitest'
+import {
+	autoOutputAsset,
+	compatibleOutputKinds,
+	generatePipelineDraft,
+	PIPELINE_OUTPUT_KINDS
+} from './pipelineTemplates'
+
+// The `data_test` output kind scaffolds a *custom* (singular) data test: a
+// standalone DuckDB script referenced from a materialize script's
+// `-- data_test <path>` line. It must be a single SELECT that reads the
+// freshly-materialized target through the internal `_wm_target` schema — the
+// two rules the backend's self-teaching errors enforce.
+describe('data_test scaffold', () => {
+	it('is a DuckDB-only output kind exposed in the picker', () => {
+		expect(compatibleOutputKinds('duckdb')).toContain('data_test')
+		expect(compatibleOutputKinds('python3')).not.toContain('data_test')
+		expect(PIPELINE_OUTPUT_KINDS.map((k) => k.id)).toContain('data_test')
+	})
+
+	it('produces no output asset (it asserts against an existing target)', () => {
+		expect(autoOutputAsset('data_test', 'folder', 'duckdb')).toBeUndefined()
+	})
+
+	it('scaffolds a single SELECT against `_wm_target.<table>`', () => {
+		const src = generatePipelineDraft({
+			language: 'duckdb',
+			outputKind: 'data_test',
+			triggers: []
+		})
+		// starter body is a single SELECT against the internal target alias.
+		expect(src).toContain('SELECT * FROM _wm_target.your_table WHERE your_condition;')
+		// exactly one SQL statement (single SELECT) — count statement lines, not
+		// the word "SELECT" that also appears in the guidance comment.
+		const stmtLines = src.split('\n').filter((l) => /^\s*SELECT\b/i.test(l))
+		expect(stmtLines).toHaveLength(1)
+		// no `-- materialize <uri>` output annotation — a data test declares no
+		// asset (the word still appears in the guidance comment, which is fine).
+		expect(src).not.toMatch(/^--\s*materialize\s/m)
+		// teaches how to wire it up + the offending-rows convention.
+		expect(src).toContain('-- data_test <this-script-path>')
+		expect(src).toContain('offending rows')
+	})
+
+	it('seeds the table name from an upstream ducklake asset when present', () => {
+		const src = generatePipelineDraft({
+			language: 'duckdb',
+			outputKind: 'data_test',
+			input: { kind: 'ducklake', path: 'analytics/orders' },
+			triggers: []
+		})
+		expect(src).toContain('SELECT * FROM _wm_target.orders WHERE your_condition;')
+		// no ATTACH of the input — the runtime attaches the target as `_wm_target`.
+		expect(src).not.toContain('ATTACH')
+	})
+})

frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts

@@ -17,6 +17,7 @@ export type PipelineOutputKind =
 	| 'datatable'
 	| 'ducklake'
 	| 'materialize'
+	| 'data_test'
 	| 's3_parquet'
 	| 's3_object'
 	| 'macros'
@@ -37,6 +38,11 @@ export const PIPELINE_OUTPUT_KINDS: PipelineOutputKindMeta[] = [
 		label: 'Materialized table',
 		description: 'Managed DuckLake table — idempotent, versioned, tracked'
 	},
+	{
+		id: 'data_test',
+		label: 'Data test',
+		description: 'Custom assertion — a single SELECT returning offending rows (empty = pass)'
+	},
 	{
 		id: 'datatable',
 		label: 'Data table',
@@ -83,7 +89,16 @@ const LANG_COMPATIBILITY: Record<ScriptLang, PipelineOutputKind[]> = {
 	// single SELECT. The Python/TS `wmll.ducklake` helper currently takes a SQL
 	// SELECT (not in-memory rows), so a polyglot managed materialize is a
 	// separate follow-up — those langs keep the `ducklake` raw-write kind.
-	duckdb: ['materialize', 'datatable', 'ducklake', 's3_parquet', 's3_object', 'macros', 'none'],
+	duckdb: [
+		'materialize',
+		'data_test',
+		'datatable',
+		'ducklake',
+		's3_parquet',
+		's3_object',
+		'macros',
+		'none'
+	],
 	postgresql: ['datatable', 'none'],
 	mysql: ['none'],
 	mssql: ['none'],
@@ -183,8 +198,10 @@ export function autoOutputAsset(
 			}
 		}
 		// A macro library produces no asset — its "output" is the registry
-		// entries the deploy records.
+		// entries the deploy records. A custom data test produces no asset
+		// either — it asserts against an existing materialized target.
 		case 'macros':
+		case 'data_test':
 		case 'none':
 			return undefined
 	}
@@ -316,6 +333,17 @@ export type TemplateContext = {
 function header(ctx: TemplateContext): string {
 	const { language, triggers, output, outputKind } = ctx
 	const p = commentPrefix(language)
+	// A custom data test is a standalone script, not a graph node that produces
+	// an asset — so it gets no `// pipeline` / output annotation. Instead, tell
+	// the author how to wire it up (the `data_test` reference) and the two rules
+	// that aren't obvious: single SELECT, returning the offending rows.
+	if (outputKind === 'data_test') {
+		return [
+			`${p} Custom data test — reference it from a materialize script with \`${p} data_test <this-script-path>\`.`,
+			`${p} It must be a single SELECT returning the offending rows; the run fails if any row comes back.`,
+			''
+		].join('\n')
+	}
 	const lines = triggers.map((t) => {
 		switch (t.kind) {
 			case 'asset':
@@ -544,6 +572,22 @@ function bodyPython(ctx: TemplateContext): string {
 function bodyDuckdb(ctx: TemplateContext): string {
 	const { input, output, outputKind } = ctx
 	const dataUpload = isDataUpload(ctx.triggers)
+	if (outputKind === 'data_test') {
+		// Standalone custom data test: it reads ONLY the freshly-materialized
+		// target, which the runtime attaches under the internal `_wm_target`
+		// schema — so it emits no ATTACH / input load of its own. When the test
+		// was created off a ducklake asset, seed its table name; otherwise a
+		// clear placeholder. This exact shape (single SELECT vs `_wm_target`) is
+		// what the backend's self-teaching errors ask for.
+		const testTable = input?.kind === 'ducklake' ? catalogTableRef(input.path) : 'your_table'
+		return [
+			'',
+			`-- Return the rows that VIOLATE your assertion; an empty result means the test passes.`,
+			`-- \`_wm_target\` is the freshly-materialized target, attached by the runtime.`,
+			`SELECT * FROM _wm_target.${testTable} WHERE your_condition;`,
+			''
+		].join('\n')
+	}
 	const lines: string[] = []
 	if (dataUpload) {
 		// `(s3object)` param declaration → the run form renders the S3 picker