mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-18 16:02:10 +00:00
Ruben Fiszel
684 lines
26 KiB
TypeScript
684 lines
26 KiB
TypeScript
import { getEnv, getWorkspace, workerHasInternalServer } from "./client";
|
|
import { OpenAPI } from "./core/OpenAPI";
|
|
import { JobService } from "./services.gen";
|
|
|
|
type ResultCollection =
|
|
| "last_statement_all_rows"
|
|
| "last_statement_first_row"
|
|
| "last_statement_all_rows_scalar"
|
|
| "last_statement_first_row_scalar"
|
|
| "all_statements_all_rows"
|
|
| "all_statements_first_row"
|
|
| "all_statements_all_rows_scalar"
|
|
| "all_statements_first_row_scalar"
|
|
| "legacy";
|
|
|
|
type FetchParams<ResultCollectionT extends ResultCollection> = {
|
|
resultCollection?: ResultCollectionT;
|
|
};
|
|
|
|
type SqlResult<
|
|
T,
|
|
ResultCollectionT extends ResultCollection
|
|
> = ResultCollectionT extends "last_statement_first_row"
|
|
? T | null
|
|
: ResultCollectionT extends "all_statements_first_row"
|
|
? T[]
|
|
: ResultCollectionT extends "last_statement_all_rows"
|
|
? T[]
|
|
: ResultCollectionT extends "all_statements_all_rows"
|
|
? T[][]
|
|
: ResultCollectionT extends "last_statement_all_rows_scalar"
|
|
? T[keyof T][]
|
|
: ResultCollectionT extends "all_statements_all_rows_scalar"
|
|
? T[keyof T][][]
|
|
: ResultCollectionT extends "last_statement_first_row_scalar"
|
|
? T[keyof T] | null
|
|
: ResultCollectionT extends "all_statements_first_row_scalar"
|
|
? T[keyof T][]
|
|
: unknown;
|
|
/**
|
|
* SQL statement object with query content, arguments, and execution methods
|
|
*/
|
|
export type SqlStatement<T> = {
|
|
/** Raw SQL content with formatted arguments */
|
|
content: string;
|
|
|
|
/** Argument values keyed by parameter name */
|
|
args: Record<string, any>;
|
|
|
|
/**
|
|
* Execute the SQL query and return results
|
|
* @param params - Optional parameters including result collection mode
|
|
* @returns Query results based on the result collection mode
|
|
*/
|
|
fetch<ResultCollectionT extends ResultCollection = "last_statement_all_rows">(
|
|
params?: FetchParams<ResultCollectionT | ResultCollection> // The union is for auto-completion
|
|
): Promise<SqlResult<T, ResultCollectionT>>;
|
|
|
|
/**
|
|
* Execute the SQL query and return only the first row
|
|
* @param params - Optional parameters
|
|
* @returns First row of the query result
|
|
*/
|
|
fetchOne(
|
|
params?: Omit<FetchParams<"last_statement_first_row">, "resultCollection">
|
|
): Promise<SqlResult<T, "last_statement_first_row">>;
|
|
|
|
/**
|
|
* Execute the SQL query and return only the first row as a scalar value
|
|
* @param params - Optional parameters
|
|
* @returns First row of the query result
|
|
*/
|
|
fetchOneScalar(
|
|
params?: Omit<
|
|
FetchParams<"last_statement_first_row_scalar">,
|
|
"resultCollection"
|
|
>
|
|
): Promise<SqlResult<T, "last_statement_first_row_scalar">>;
|
|
|
|
/**
|
|
* Execute the SQL query without fetching rows
|
|
* @param params - Optional parameters
|
|
*/
|
|
execute(
|
|
params?: Omit<FetchParams<"last_statement_first_row">, "resultCollection">
|
|
): Promise<void>;
|
|
};
|
|
|
|
/**
|
|
* Wrapper for raw SQL fragments that should be inlined without parameterization.
|
|
* Created via `sql.raw(value)`.
|
|
*/
|
|
export class RawSql {
|
|
readonly __brand = "RawSql" as const;
|
|
constructor(public readonly value: string) { }
|
|
}
|
|
|
|
/**
|
|
* Template tag function for creating SQL statements with parameterized values
|
|
*/
|
|
export interface SqlTemplateFunction {
|
|
<T = any>(strings: TemplateStringsArray, ...values: any[]): SqlStatement<T>;
|
|
/** Create a raw SQL fragment that will be inlined without parameterization */
|
|
raw(value: string): RawSql;
|
|
}
|
|
|
|
export interface DatatableSqlTemplateFunction extends SqlTemplateFunction {
|
|
query<T = any>(sql: string, ...params: any[]): SqlStatement<T>;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Provider interface — captures what differs between datatable and ducklake
|
|
// ---------------------------------------------------------------------------
|
|
|
|
interface SqlProvider {
|
|
formatArgDecl(argNum: number, argType: string): string;
|
|
formatArgUsage(
|
|
argNum: number,
|
|
explicitType: string | undefined,
|
|
inferredType: string
|
|
): string;
|
|
preamble(): string;
|
|
language: "postgresql" | "duckdb";
|
|
extraArgs: Record<string, any>;
|
|
providerName: string;
|
|
}
|
|
|
|
function datatableProvider(name: string, schema?: string): SqlProvider {
|
|
return {
|
|
providerName: "datatable",
|
|
language: "postgresql",
|
|
extraArgs: { database: `datatable://${name}` },
|
|
formatArgDecl: (argNum) => `-- $${argNum} arg${argNum}`,
|
|
formatArgUsage: (argNum, explicitType, inferredType) =>
|
|
explicitType !== undefined
|
|
? `$${argNum}`
|
|
: `$${argNum}::${inferredType}`,
|
|
preamble: () => (schema ? `SET search_path TO "${schema}";\n` : ""),
|
|
};
|
|
}
|
|
|
|
function ducklakeProvider(name: string, schema?: string): SqlProvider {
|
|
return {
|
|
providerName: "ducklake",
|
|
language: "duckdb",
|
|
extraArgs: {},
|
|
formatArgDecl: (argNum, argType) => `-- $arg${argNum} (${argType})`,
|
|
formatArgUsage: (argNum) => `$arg${argNum}`,
|
|
// `USE dl."schema"` sets the active schema so unqualified tables resolve there.
|
|
preamble: () =>
|
|
`ATTACH 'ducklake://${name}' AS dl;USE dl${schema ? `."${schema}"` : ""};\n`,
|
|
};
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Shared template function builder
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// Build a ready-to-execute SqlStatement. Used by both the template-tag
|
|
// path (which builds `content` from strings/values) and `.query()` (which
|
|
// gets a hand-written SQL string with positional placeholders).
|
|
function buildSqlStatement(
|
|
provider: SqlProvider,
|
|
content: string,
|
|
contentBody: string,
|
|
args: Record<string, any>
|
|
): SqlStatement<any> {
|
|
async function fetch<ResultCollectionT extends ResultCollection>({
|
|
resultCollection,
|
|
}: FetchParams<ResultCollectionT> = {}) {
|
|
let finalContent = content;
|
|
if (resultCollection)
|
|
finalContent = `-- result_collection=${resultCollection}\n${finalContent}`;
|
|
try {
|
|
let result;
|
|
if (workerHasInternalServer()) {
|
|
result = await JobService.runScriptPreviewInline({
|
|
workspace: getWorkspace(),
|
|
requestBody: { args, content: finalContent, language: provider.language },
|
|
});
|
|
} else {
|
|
result = await JobService.runScriptPreviewAndWaitResult({
|
|
workspace: getWorkspace(),
|
|
requestBody: { args, content: finalContent, language: provider.language },
|
|
});
|
|
}
|
|
return result as SqlResult<any, ResultCollectionT>;
|
|
} catch (e: any) {
|
|
let err = e;
|
|
if (
|
|
e &&
|
|
typeof e.body == "string" &&
|
|
e.statusText == "Internal Server Error"
|
|
) {
|
|
let body = e.body;
|
|
if (body.startsWith("Internal:")) body = body.slice(9).trim();
|
|
if (body.startsWith("Error:")) body = body.slice(6).trim();
|
|
if (body.startsWith("datatable")) body = body.slice(9).trim();
|
|
err = Error(`${provider.providerName} ${body}`);
|
|
err.query = contentBody;
|
|
err.request = e.request;
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
return {
|
|
content,
|
|
args,
|
|
fetch,
|
|
fetchOne: (params) =>
|
|
fetch({ ...params, resultCollection: "last_statement_first_row" }),
|
|
fetchOneScalar: (params) =>
|
|
fetch({
|
|
...params,
|
|
resultCollection: "last_statement_first_row_scalar",
|
|
}),
|
|
execute: (params) => fetch(params),
|
|
} satisfies SqlStatement<any>;
|
|
}
|
|
|
|
// JSON-encode a JS value into something the executor can deserialize. The
|
|
// JSON.stringify-friendly representation of a JS value before sending it to
|
|
// the executor:
|
|
// - `bigint` → string. JSON.stringify on a bigint throws; the
|
|
// executor accepts numeric strings into BIGINT
|
|
// slots via `Value::String → INT8`.
|
|
// - `Date` → ISO-8601 string. inferSqlType maps these to
|
|
// `TIMESTAMPTZ`; the executor's `Value::String`
|
|
// arm parses ISO strings into `chrono::DateTime`.
|
|
// - non-finite `number` → string ("NaN" / "Infinity" / "-Infinity").
|
|
// JSON.stringify renders these as `null`, which
|
|
// silently became NULL in the database. The
|
|
// executor accepts these literals via
|
|
// `Value::String → FLOAT8` (`f64::from_str`).
|
|
// - everything else → passed through unchanged.
|
|
function serializeArgValue(v: any): any {
|
|
if (typeof v === "bigint") return v.toString();
|
|
if (v instanceof Date) return v.toISOString();
|
|
if (typeof v === "number" && !Number.isFinite(v)) {
|
|
if (Number.isNaN(v)) return "NaN";
|
|
return v > 0 ? "Infinity" : "-Infinity";
|
|
}
|
|
return v;
|
|
}
|
|
|
|
function buildSqlTemplateFunction(provider: SqlProvider): SqlTemplateFunction {
|
|
let sqlFn = ((strings: TemplateStringsArray, ...values: any[]) => {
|
|
// Separate raw vs parameterized values, assigning arg indices only to params
|
|
let argIndex = 0;
|
|
const valueInfos = values.map((v, i) => {
|
|
if (v instanceof RawSql)
|
|
return { raw: true as const, value: v.value, originalIndex: i };
|
|
argIndex++;
|
|
return {
|
|
raw: false as const,
|
|
value: v,
|
|
originalIndex: i,
|
|
argNum: argIndex,
|
|
};
|
|
});
|
|
|
|
// Arg declarations (SQL comments consumed by the executor)
|
|
let argDecls = valueInfos
|
|
.filter((info): info is Extract<(typeof valueInfos)[number], { raw: false }> => !info.raw)
|
|
.map((info) => {
|
|
let argType =
|
|
parseTypeAnnotation(
|
|
strings[info.originalIndex],
|
|
strings[info.originalIndex + 1]
|
|
) || inferSqlType(info.value);
|
|
return provider.formatArgDecl(info.argNum, argType);
|
|
});
|
|
|
|
let content = argDecls.length ? argDecls.join("\n") + "\n" : "";
|
|
content += provider.preamble();
|
|
|
|
// SQL body — inline raw values, reference params via provider syntax
|
|
let contentBody = "";
|
|
for (let i = 0; i < strings.length; i++) {
|
|
contentBody += strings[i];
|
|
if (i < valueInfos.length) {
|
|
let info = valueInfos[i];
|
|
if (info.raw) {
|
|
contentBody += info.value;
|
|
} else {
|
|
let explicitType = parseTypeAnnotation(
|
|
strings[info.originalIndex],
|
|
strings[info.originalIndex + 1]
|
|
);
|
|
let inferredType = inferSqlType(info.value);
|
|
contentBody += provider.formatArgUsage(
|
|
info.argNum,
|
|
explicitType,
|
|
inferredType
|
|
);
|
|
}
|
|
}
|
|
}
|
|
content += contentBody;
|
|
|
|
const args = {
|
|
...Object.fromEntries(
|
|
valueInfos
|
|
.filter((info): info is Extract<(typeof valueInfos)[number], { raw: false }> => !info.raw)
|
|
.map((info) => [`arg${info.argNum}`, serializeArgValue(info.value)])
|
|
),
|
|
...provider.extraArgs,
|
|
};
|
|
|
|
return buildSqlStatement(provider, content, contentBody, args);
|
|
}) as SqlTemplateFunction;
|
|
|
|
sqlFn.raw = (value: string) => new RawSql(value);
|
|
return sqlFn;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Public API
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Create a SQL template function for PostgreSQL/datatable queries
|
|
* @param name - Database/datatable name (default: "main")
|
|
* @returns SQL template function for building parameterized queries
|
|
* @example
|
|
* let sql = wmill.datatable()
|
|
* let name = 'Robin'
|
|
* let age = 21
|
|
* await sql`
|
|
* SELECT * FROM friends
|
|
* WHERE name = ${name} AND age = ${age}::int
|
|
* `.fetch()
|
|
*/
|
|
export function datatable(name: string = "main"): DatatableSqlTemplateFunction {
|
|
let { name: n, schema } = parseName(name);
|
|
let provider = datatableProvider(n, schema);
|
|
let sqlFn = buildSqlTemplateFunction(provider) as DatatableSqlTemplateFunction;
|
|
// `.query(sql, ...params)` is for SQL strings that already contain
|
|
// positional placeholders ($1, $2, ...). We DON'T go through the template
|
|
// builder here — that would re-emit each value as `$N::TYPE` and append
|
|
// them after the user's literal SQL, which is the bug previous versions of
|
|
// this method shipped. Instead we build the executor-shaped content
|
|
// directly: a `-- $N argN (TYPE)` declaration block (the parser picks
|
|
// these up as explicitly typed args) followed by the user's SQL verbatim.
|
|
// Note: we hand-roll the decl format here rather than calling
|
|
// `provider.formatArgDecl`, because the datatable formatter intentionally
|
|
// omits the type (the template-tag path emits `$N::TYPE` inline instead);
|
|
// for `.query()` we have no inline cast to fall back on.
|
|
sqlFn.query = (sqlString: string, ...params: any[]) => {
|
|
let argDecls = params
|
|
.map((v, i) => `-- $${i + 1} arg${i + 1} (${inferSqlType(v)})`)
|
|
.join("\n");
|
|
let contentBody = sqlString;
|
|
let content =
|
|
(argDecls ? argDecls + "\n" : "") + provider.preamble() + sqlString;
|
|
let args = {
|
|
...Object.fromEntries(
|
|
params.map((v, i) => [`arg${i + 1}`, serializeArgValue(v)])
|
|
),
|
|
...provider.extraArgs,
|
|
};
|
|
return buildSqlStatement(provider, content, contentBody, args);
|
|
};
|
|
return sqlFn;
|
|
}
|
|
|
|
/**
|
|
* Create a SQL template function for DuckDB/ducklake queries
|
|
* @param name - DuckDB database name, optionally with a schema as `name:schema` (default: "main")
|
|
* @returns SQL template function for building parameterized queries
|
|
* @example
|
|
* let sql = wmill.ducklake()
|
|
* let name = 'Robin'
|
|
* let age = 21
|
|
* await sql`
|
|
* SELECT * FROM friends
|
|
* WHERE name = ${name} AND age = ${age}
|
|
* `.fetch()
|
|
* @example
|
|
* // Target a specific schema within the ducklake
|
|
* let sql = wmill.ducklake("my_lake:analytics")
|
|
*/
|
|
export function ducklake(name: string = "main"): SqlTemplateFunction {
|
|
let { name: n, schema } = parseName(name);
|
|
return buildSqlTemplateFunction(ducklakeProvider(n, schema));
|
|
}
|
|
|
|
/** Options for the ducklake materialization helpers. `partition` is bound as a
|
|
* DuckDB arg (never interpolated); `selectSql`, `table`, `schema`, `uniqueKey`
|
|
* are trusted structural SQL inlined via `raw`. */
|
|
export interface DucklakeMaterializeOptions {
|
|
/** ducklake name (default "main"), optionally "name:schema". */
|
|
ducklake?: string;
|
|
/** target table within the ducklake. */
|
|
table: string;
|
|
/** the SELECT producing the rows for this slice. */
|
|
selectSql: string;
|
|
/** the partition value (bound). Omit for a whole-table materialization — no
|
|
* partition column, and replace becomes a `CREATE OR REPLACE TABLE`. */
|
|
partition?: string;
|
|
/** dedup key → upsert in slice (delete-by-key + insert); omit → replace (delete partition + insert). */
|
|
uniqueKey?: string;
|
|
/** physical partition column (default "_wm_partition"). */
|
|
partitionCol?: string;
|
|
}
|
|
|
|
/** Idempotently materialize `selectSql` into a ducklake table for one
|
|
* partition (or the whole table when `partition` is omitted) — the client-side
|
|
* equivalent of the `// materialize` engine.
|
|
* With `uniqueKey` it upserts the slice (delete-by-key + insert); otherwise it
|
|
* replaces it (whole table → `CREATE OR REPLACE`; partition → delete + insert).
|
|
* Safe to re-run for the same partition (backfill / failure-recovery).
|
|
*
|
|
* Returns a lazy statement — call `.execute()` to run it:
|
|
* `await wmill.upsertPartition({ table, selectSql, partition }).execute()`. */
|
|
export function upsertPartition(opts: DucklakeMaterializeOptions): SqlStatement<any> {
|
|
return finishMaterialize(buildUpsertStatement(opts), opts);
|
|
}
|
|
function buildUpsertStatement(opts: DucklakeMaterializeOptions): SqlStatement<any> {
|
|
let { name: n, schema } = parseName(opts.ducklake ?? "main");
|
|
let sql = buildSqlTemplateFunction(ducklakeProvider(n, schema));
|
|
let pcol = sql.raw(opts.partitionCol ?? "_wm_partition");
|
|
let t = sql.raw(`dl.${opts.table}`);
|
|
let body = sql.raw(opts.selectSql);
|
|
// Whole-table (no partition): no partition column. Replace rebuilds the table
|
|
// with CREATE OR REPLACE (handles schema changes); merge upserts by key.
|
|
if (opts.partition === undefined) {
|
|
if (opts.uniqueKey) {
|
|
let uk = sql.raw(opts.uniqueKey);
|
|
return sql`CREATE TABLE IF NOT EXISTS ${t} AS SELECT * FROM (${body}) WHERE false;
|
|
BEGIN TRANSACTION;
|
|
DELETE FROM ${t} WHERE ${uk} IN (SELECT ${uk} FROM (${body}));
|
|
INSERT INTO ${t} SELECT * FROM (${body});
|
|
COMMIT;`;
|
|
}
|
|
return sql`CREATE OR REPLACE TABLE ${t} AS SELECT * FROM (${body});`;
|
|
}
|
|
if (opts.uniqueKey) {
|
|
let uk = sql.raw(opts.uniqueKey);
|
|
// Upsert via delete-by-key + insert (not MERGE — DuckLake's MERGE fails
|
|
// writing the first rows of a fresh partition).
|
|
return sql`CREATE TABLE IF NOT EXISTS ${t} AS SELECT *, CAST(NULL AS VARCHAR) AS ${pcol} FROM (${body}) WHERE false;
|
|
ALTER TABLE ${t} SET PARTITIONED BY (${pcol});
|
|
BEGIN TRANSACTION;
|
|
DELETE FROM ${t} WHERE ${pcol} = ${opts.partition} AND ${uk} IN (SELECT ${uk} FROM (${body}));
|
|
INSERT INTO ${t} SELECT *, ${opts.partition} AS ${pcol} FROM (${body});
|
|
COMMIT;`;
|
|
}
|
|
return sql`CREATE TABLE IF NOT EXISTS ${t} AS SELECT *, CAST(NULL AS VARCHAR) AS ${pcol} FROM (${body}) WHERE false;
|
|
ALTER TABLE ${t} SET PARTITIONED BY (${pcol});
|
|
BEGIN TRANSACTION;
|
|
DELETE FROM ${t} WHERE ${pcol} = ${opts.partition};
|
|
INSERT INTO ${t} SELECT *, ${opts.partition} AS ${pcol} FROM (${body});
|
|
COMMIT;`;
|
|
}
|
|
|
|
/** INSERT-only materialization (no dedup/replace) for append-only tables.
|
|
* Re-running the same partition duplicates rows — use only for immutable
|
|
* event-log sources.
|
|
*
|
|
* Returns a lazy statement — call `.execute()` to run it:
|
|
* `await wmill.appendPartition({ table, selectSql, partition }).execute()`. */
|
|
export function appendPartition(
|
|
opts: Omit<DucklakeMaterializeOptions, "uniqueKey">,
|
|
): SqlStatement<any> {
|
|
return finishMaterialize(buildAppendStatement(opts), opts);
|
|
}
|
|
function buildAppendStatement(
|
|
opts: Omit<DucklakeMaterializeOptions, "uniqueKey">,
|
|
): SqlStatement<any> {
|
|
let { name: n, schema } = parseName(opts.ducklake ?? "main");
|
|
let sql = buildSqlTemplateFunction(ducklakeProvider(n, schema));
|
|
let pcol = sql.raw(opts.partitionCol ?? "_wm_partition");
|
|
let t = sql.raw(`dl.${opts.table}`);
|
|
let body = sql.raw(opts.selectSql);
|
|
// Whole-table (no partition): insert into the bare table, no partition column.
|
|
if (opts.partition === undefined) {
|
|
return sql`CREATE TABLE IF NOT EXISTS ${t} AS SELECT * FROM (${body}) WHERE false;
|
|
INSERT INTO ${t} SELECT * FROM (${body});`;
|
|
}
|
|
return sql`CREATE TABLE IF NOT EXISTS ${t} AS SELECT *, CAST(NULL AS VARCHAR) AS ${pcol} FROM (${body}) WHERE false;
|
|
ALTER TABLE ${t} SET PARTITIONED BY (${pcol});
|
|
INSERT INTO ${t} SELECT *, ${opts.partition} AS ${pcol} FROM (${body});`;
|
|
}
|
|
|
|
// In pipeline context (WM_PIPELINE), wrap a materialize statement so a
|
|
// successful run captures the slice's row count + snapshot and records
|
|
// materialized_partition state — making SDK materializations appear in the grid
|
|
// like `// materialize` ones. Outside a pipeline it's a passthrough (no record).
|
|
function finishMaterialize(
|
|
stmt: SqlStatement<any>,
|
|
opts: Pick<DucklakeMaterializeOptions, "ducklake" | "table" | "partition" | "partitionCol">,
|
|
): SqlStatement<any> {
|
|
if (getEnv("WM_PIPELINE") !== "true") return stmt;
|
|
let { name: n, schema } = parseName(opts.ducklake ?? "main");
|
|
let sql = buildSqlTemplateFunction(ducklakeProvider(n, schema));
|
|
let t = sql.raw(`dl.${opts.table}`);
|
|
let pcol = opts.partitionCol ?? "_wm_partition";
|
|
let where =
|
|
opts.partition !== undefined
|
|
? sql.raw(`WHERE ${pcol} = '${String(opts.partition).replace(/'/g, "''")}'`)
|
|
: sql.raw("");
|
|
let summary = sql`SELECT (SELECT count(*) FROM ${t} ${where}) AS rows, (SELECT max(snapshot_id) FROM ducklake_snapshots('dl')) AS snapshot_id`;
|
|
// Asset path mirrors the `// materialize` engine: <lake>/<schema>.<table> for
|
|
// an explicit schema, else <lake>/<table> — so the grid lookup matches and
|
|
// distinct schemas don't collide under one state key.
|
|
let assetPath = schema ? `${n}/${schema}.${opts.table}` : `${n}/${opts.table}`;
|
|
let partition = opts.partition ?? "";
|
|
let run = async () => {
|
|
try {
|
|
await stmt.execute();
|
|
} catch (e) {
|
|
await recordMaterialization(assetPath, partition, "failed", null, null, String(e));
|
|
throw e;
|
|
}
|
|
let snapshot_id: number | null = null;
|
|
let row_count: number | null = null;
|
|
try {
|
|
let s: any = await summary.fetchOne();
|
|
snapshot_id = s?.snapshot_id ?? null;
|
|
row_count = s?.rows ?? null;
|
|
} catch {
|
|
/* summary read is best-effort */
|
|
}
|
|
await recordMaterialization(assetPath, partition, "materialized", snapshot_id, row_count, null);
|
|
};
|
|
return {
|
|
...stmt,
|
|
execute: (() => run()) as any,
|
|
fetch: (() => run()) as any,
|
|
fetchOne: (() => run()) as any,
|
|
fetchOneScalar: (() => run()) as any,
|
|
};
|
|
}
|
|
|
|
async function recordMaterialization(
|
|
assetPath: string,
|
|
partition: string,
|
|
status: string,
|
|
snapshot_id: number | null,
|
|
row_count: number | null,
|
|
error: string | null,
|
|
): Promise<void> {
|
|
try {
|
|
await fetch(`${OpenAPI.BASE}/w/${getWorkspace()}/assets/record_materialization`, {
|
|
method: "POST",
|
|
headers: {
|
|
"content-type": "application/json",
|
|
authorization: `Bearer ${OpenAPI.TOKEN as string}`,
|
|
},
|
|
body: JSON.stringify({
|
|
asset_kind: "ducklake",
|
|
asset_path: assetPath,
|
|
partition,
|
|
status,
|
|
snapshot_id,
|
|
row_count,
|
|
job_id: getEnv("WM_JOB_ID") ?? null,
|
|
error,
|
|
}),
|
|
});
|
|
} catch {
|
|
// best-effort; never fail the user's materialization
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Utilities
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// DuckDB executor requires explicit argument types at declaration
|
|
// And postgres at argument usage.
|
|
// These types exist in both DuckDB and Postgres
|
|
// Check that the types exist if you plan to extend this function for other SQL engines.
|
|
function inferSqlType(value: any): string {
|
|
if (typeof value === "bigint") return "BIGINT";
|
|
if (typeof value === "number") {
|
|
if (Number.isInteger(value)) return "BIGINT";
|
|
return "DOUBLE PRECISION";
|
|
} else if (value === null || value === undefined) {
|
|
return "TEXT";
|
|
} else if (typeof value === "string") {
|
|
return "TEXT";
|
|
} else if (Array.isArray(value)) {
|
|
// Homogeneous-primitive arrays auto-tag as `TYPE[]` so that values like
|
|
// `${[1,2,3]}` against an `int[]` column work without an explicit
|
|
// `${arr}::int[]` cast. For non-homogeneous or nested arrays we fall
|
|
// back to JSON, which works for jsonb columns.
|
|
return inferSqlArrayType(value);
|
|
} else if (value instanceof Date) {
|
|
// JS `Date` carries an absolute instant in UTC; map to TIMESTAMPTZ so
|
|
// `${someDate}` works against a `timestamptz` column without the user
|
|
// needing an explicit cast. Without this the typeof check above falls
|
|
// through to "object" → JSON, which only works by accident via
|
|
// PG's `json → text → timestamptz` implicit cast chain.
|
|
return "TIMESTAMPTZ";
|
|
} else if (typeof value === "object") {
|
|
return "JSON";
|
|
} else if (typeof value === "boolean") {
|
|
return "BOOLEAN";
|
|
} else {
|
|
return "TEXT";
|
|
}
|
|
}
|
|
|
|
function inferSqlArrayType(value: any[]): string {
|
|
if (value.length === 0) return "JSON";
|
|
// Detect a single shared scalar JS type across all elements. Mixed types
|
|
// or any non-primitive element forces the JSON fallback.
|
|
let scalarType: string | undefined = undefined;
|
|
for (const elem of value) {
|
|
let elemType: string;
|
|
if (typeof elem === "bigint") elemType = "BIGINT";
|
|
else if (typeof elem === "number")
|
|
elemType = Number.isInteger(elem) ? "BIGINT" : "DOUBLE PRECISION";
|
|
else if (typeof elem === "string") elemType = "TEXT";
|
|
else if (typeof elem === "boolean") elemType = "BOOLEAN";
|
|
else return "JSON";
|
|
if (scalarType === undefined) scalarType = elemType;
|
|
else if (scalarType === "BIGINT" && elemType === "DOUBLE PRECISION")
|
|
scalarType = "DOUBLE PRECISION";
|
|
else if (scalarType === "DOUBLE PRECISION" && elemType === "BIGINT") {
|
|
// already widened
|
|
} else if (scalarType !== elemType) {
|
|
return "JSON";
|
|
}
|
|
}
|
|
return `${scalarType}[]`;
|
|
}
|
|
|
|
// The goal is to detect if the user added a type annotation manually
|
|
//
|
|
// untyped : sql`SELECT ${x} = 0` => ['SELECT ', ' = 0']
|
|
// typed : sql`SELECT ${x}::int = 0` => ['SELECT ', '::int = 0']
|
|
// typed : sql`SELECT CAST ( ${x} AS int ) = 0` => ['SELECT CAST ( ', ' AS int ) = 0']
|
|
//
|
|
// Caveat: the returned string is only meaningful as a *presence* signal —
|
|
// the only consumer (`formatArgUsage`) just checks `explicitType !== undefined`
|
|
// to decide whether to emit `$N` (user already wrote a cast) vs `$N::TYPE`
|
|
// (SDK injects the inferred cast). The returned string itself can be
|
|
// imprecise — e.g. `${x}::DOUBLE PRECISION` returns `"DOUBLE"` (split on
|
|
// whitespace), and `CAST(${x} AS int)` returns `"int)"` (no paren stripping).
|
|
// Don't rely on the returned string as a parsed PG type; only on whether
|
|
// it's defined.
|
|
function parseTypeAnnotation(
|
|
prevTemplateString: string | undefined,
|
|
nextTemplateString: string | undefined
|
|
): string | undefined {
|
|
if (!nextTemplateString) return;
|
|
nextTemplateString = nextTemplateString.trimStart();
|
|
if (nextTemplateString.startsWith("::")) {
|
|
return nextTemplateString.substring(2).trimStart().split(/\s+/)[0];
|
|
}
|
|
prevTemplateString = prevTemplateString?.trimEnd();
|
|
if (
|
|
prevTemplateString?.endsWith("(") &&
|
|
prevTemplateString
|
|
.substring(0, prevTemplateString.length - 1)
|
|
.trim()
|
|
.toUpperCase()
|
|
.endsWith("CAST") &&
|
|
nextTemplateString.toUpperCase().startsWith("AS ")
|
|
) {
|
|
return nextTemplateString.substring(2).trimStart().split(/\s+/)[0];
|
|
}
|
|
}
|
|
|
|
function parseName(name: string | undefined): {
|
|
name: string;
|
|
schema?: string;
|
|
} {
|
|
if (!name) return { name: "main" };
|
|
let [assetName, schemaName] = name.split(":");
|
|
if (schemaName) {
|
|
return {
|
|
name: assetName || "main",
|
|
schema: schemaName,
|
|
};
|
|
} else {
|
|
return { name };
|
|
}
|
|
}
|