use super::{FnDescriptor, FnVolatility, ImplStrategy, Param, PylonFnDef, PylonType, SqlLanguage};
fn pg_type(ty: &PylonType) -> String {
use PylonType::*;
match ty {
Str => "text".into(),
Bool => "bool".into(),
Int16 => "int2".into(),
Int32 => "int4".into(),
Int64 => "int8".into(),
Float32 => "float4".into(),
Float64 => "float8".into(),
Decimal | BigInt => "numeric".into(),
Uuid => "uuid".into(),
Json => "jsonb".into(),
Bytes => "bytea".into(),
Datetime => "timestamptz".into(),
Duration | RelativeDuration | DateDuration => "interval".into(),
LocalDatetime => "timestamp".into(),
LocalDate => "date".into(),
LocalTime => "time".into(),
Vector => "vector".into(),
Geometry => "geometry".into(),
Geography => "geography".into(),
Box2D => "box2d".into(),
Box3D => "box3d".into(),
Any | AnyOrderable | AnyPoint => "anyelement".into(),
Array(inner) => match inner.as_ref() {
Any | AnyOrderable | AnyPoint => "anyarray".into(),
other => format!("{}[]", pg_type(other)),
},
Set(inner) => match inner.as_ref() {
Any | AnyOrderable | AnyPoint => "anyarray".into(),
other => format!("{}[]", pg_type(other)),
},
Optional(inner) => pg_type(inner),
Range(inner) => match inner.as_ref() {
Any | AnyOrderable | AnyPoint => "anyrange".into(),
other => format!("{}range", pg_type(other)),
},
Multirange(inner) => match inner.as_ref() {
Any | AnyOrderable | AnyPoint => "anymultirange".into(),
other => format!("{}multirange", pg_type(other)),
},
Tuple(_) => panic!("Tuple cannot appear as a PG function parameter type"),
}
}
fn pg_returns(ty: &PylonType) -> String {
use PylonType::*;
match ty {
Set(inner) => match inner.as_ref() {
Tuple(_) => panic!("TABLE returns must use PylonFnDef::returns_override"),
other => format!("SETOF {}", pg_type(other)),
},
Optional(inner) => pg_type(inner),
other => pg_type(other),
}
}
fn pg_params(params: &[Param]) -> String {
if params.is_empty() {
return String::new();
}
let last = params.len() - 1;
params
.iter()
.enumerate()
.map(|(i, p)| {
let arr_ty = match &p.ty {
PylonType::Any | PylonType::AnyOrderable | PylonType::AnyPoint => "anyarray".into(),
other => format!("{}[]", pg_type(other)),
};
if p.variadic && i == last {
format!("VARIADIC {} {}", p.name, arr_ty)
} else if p.variadic {
format!("{} {}", p.name, arr_ty)
} else {
format!("{} {}", p.name, pg_type(&p.ty))
}
})
.collect::<Vec<_>>()
.join(", ")
}
fn render_function(desc: &FnDescriptor, def: &PylonFnDef) -> String {
let params = pg_params(&desc.params);
let returns = def
.returns_override
.map(|s| s.to_owned())
.unwrap_or_else(|| pg_returns(&desc.return_type));
let lang = match def.language {
SqlLanguage::Sql => "sql",
SqlLanguage::PlPgSql => "plpgsql",
};
let volatility = match def.volatility {
FnVolatility::Immutable => "IMMUTABLE",
FnVolatility::Stable => "STABLE",
FnVolatility::Volatile | FnVolatility::Modifying => "VOLATILE",
};
let parallel = match def.volatility {
FnVolatility::Modifying => "PARALLEL UNSAFE",
_ => "PARALLEL SAFE",
};
let strict = if def.strict { " STRICT" } else { "" };
format!(
"CREATE OR REPLACE FUNCTION _pylon.{name}({params})\n\
\tRETURNS {returns}\n\
\tLANGUAGE {lang} {volatility} {parallel}{strict}\n\
AS $$\n\
{body}\n\
$$;\n",
name = def.name,
body = def.body,
)
}
pub const INDEX_OUTBOX_DDL: &str = concat!(
"DO $$ BEGIN\n",
" CREATE TYPE _pylon.\"IndexKind\" AS ENUM ('Vector', 'OpenSearch', 'Meilisearch');\n",
"EXCEPTION WHEN duplicate_object THEN NULL; END $$;\n",
"ALTER TYPE _pylon.\"IndexKind\" ADD VALUE IF NOT EXISTS 'Meilisearch';\n",
"DO $$ BEGIN\n",
" CREATE TYPE _pylon.\"IndexOutboxStatus\" AS ENUM ('Pending', 'Processing', 'Failed');\n",
"EXCEPTION WHEN duplicate_object THEN NULL; END $$;\n\n",
"CREATE TABLE IF NOT EXISTS _pylon.\"IndexOutbox\" (\n",
" id uuid NOT NULL DEFAULT uuidv7(),\n",
" object_id uuid NOT NULL,\n",
" type_name text NOT NULL,\n",
" index_kind _pylon.\"IndexKind\" NOT NULL,\n",
" index_name text,\n",
" operation text NOT NULL DEFAULT 'index',\n",
" status _pylon.\"IndexOutboxStatus\" NOT NULL DEFAULT 'Pending',\n",
" attempts int NOT NULL DEFAULT 0,\n",
" enqueued_at timestamptz NOT NULL DEFAULT now(),\n",
" next_attempt timestamptz,\n",
" claimed_at timestamptz,\n",
" PRIMARY KEY (id),\n",
" UNIQUE NULLS NOT DISTINCT (object_id, index_kind, index_name)\n",
");\n",
"ALTER TABLE _pylon.\"IndexOutbox\" ADD COLUMN IF NOT EXISTS\n",
" operation text NOT NULL DEFAULT 'index';\n",
"ALTER TABLE _pylon.\"IndexOutbox\" ADD COLUMN IF NOT EXISTS\n",
" claimed_at timestamptz;\n\n",
"CREATE INDEX IF NOT EXISTS \"IndexOutbox_status_next_attempt\" ON _pylon.\"IndexOutbox\" (status, next_attempt)\n",
" WHERE status IN ('Pending', 'Failed');\n\n",
"CREATE OR REPLACE FUNCTION _pylon.notify_index_queue()\n",
" RETURNS trigger LANGUAGE plpgsql AS $$\n",
"BEGIN\n",
" PERFORM pg_notify('pylon_index_queue', NEW.object_id::text);\n",
" RETURN NEW;\n",
"END\n",
"$$;\n\n",
"CREATE OR REPLACE TRIGGER notify_index_queue\n",
" AFTER INSERT OR UPDATE ON _pylon.\"IndexOutbox\"\n",
" FOR EACH ROW EXECUTE FUNCTION _pylon.notify_index_queue();\n",
);
pub const SIGNAL_OUTBOX_DDL: &str = concat!(
"DO $$ BEGIN\n",
" CREATE TYPE _pylon.\"SignalOutboxStatus\" AS ENUM ('Pending', 'Processing', 'Failed');\n",
"EXCEPTION WHEN duplicate_object THEN NULL; END $$;\n\n",
"CREATE TABLE IF NOT EXISTS _pylon.\"SignalOutbox\" (\n",
" id uuid NOT NULL DEFAULT uuidv7(),\n",
" type_name text NOT NULL,\n",
" operation text NOT NULL,\n",
" old_row jsonb,\n",
" new_row jsonb,\n",
" status _pylon.\"SignalOutboxStatus\" NOT NULL DEFAULT 'Pending',\n",
" attempts int NOT NULL DEFAULT 0,\n",
" enqueued_at timestamptz NOT NULL DEFAULT now(),\n",
" next_attempt timestamptz,\n",
" PRIMARY KEY (id)\n",
");\n\n",
"CREATE INDEX IF NOT EXISTS \"SignalOutbox_status_next_attempt\" ON _pylon.\"SignalOutbox\" (status, next_attempt)\n",
" WHERE status IN ('Pending', 'Failed');\n\n",
"CREATE OR REPLACE FUNCTION _pylon.notify_signal_queue()\n",
" RETURNS trigger LANGUAGE plpgsql AS $$\n",
"BEGIN\n",
" PERFORM pg_notify('pylon_signal_queue', NEW.id::text);\n",
" RETURN NEW;\n",
"END\n",
"$$;\n\n",
"CREATE OR REPLACE TRIGGER notify_signal_queue\n",
" AFTER INSERT ON _pylon.\"SignalOutbox\"\n",
" FOR EACH ROW EXECUTE FUNCTION _pylon.notify_signal_queue();\n",
);
pub const CACHE_INVALIDATE_DDL: &str = concat!(
"CREATE OR REPLACE FUNCTION _pylon.notify_cache_invalidate()\n",
" RETURNS trigger LANGUAGE plpgsql AS $$\n",
"BEGIN\n",
" PERFORM pg_notify('pylon_cache_invalidate', TG_TABLE_SCHEMA || '.' || TG_TABLE_NAME);\n",
" RETURN NULL;\n",
"END\n",
"$$;\n",
);
pub const MIGRATION_TRACKING_DDL: &str = concat!(
"CREATE TABLE IF NOT EXISTS _pylon.\"Migrations\" (\n",
" id text PRIMARY KEY,\n",
" onto text NOT NULL,\n",
" filename text NOT NULL,\n",
" db_state jsonb NULL,\n",
" applied_at timestamptz NULL\n",
");\n",
"ALTER TABLE _pylon.\"Migrations\" ADD COLUMN IF NOT EXISTS schema_state jsonb NULL;\n\n",
"CREATE TABLE IF NOT EXISTS _pylon.\"Progress\" (\n",
" id text PRIMARY KEY,\n",
" step_index integer NOT NULL,\n",
" updated_at timestamptz NOT NULL DEFAULT now()\n",
");\n\n",
"CREATE TABLE IF NOT EXISTS _pylon.\"Schema\" (\n",
" singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),\n",
" snapshot jsonb NOT NULL,\n",
" updated_at timestamptz NOT NULL DEFAULT now()\n",
");\n\n",
"CREATE TABLE IF NOT EXISTS _pylon.\"Internal\" (\n",
" singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),\n",
" version integer NOT NULL,\n",
" updated_at timestamptz NOT NULL DEFAULT now()\n",
");\n",
"INSERT INTO _pylon.\"Internal\" (singleton, version) VALUES (true, ",
internal_schema_version_literal!(),
")\n",
" ON CONFLICT (singleton) DO UPDATE SET version = ",
internal_schema_version_literal!(),
", updated_at = now();\n",
);
pub const INTERNAL_SCHEMA_VERSION: i32 = 1;
pub const MIN_SUPPORTED_INTERNAL_VERSION: i32 = 1;
macro_rules! internal_schema_version_literal {
() => {
"1"
};
}
use internal_schema_version_literal;
pub fn export_stdlib() -> String {
let mut out = String::from("CREATE SCHEMA IF NOT EXISTS _pylon;\n\n");
out.push_str(INDEX_OUTBOX_DDL);
out.push('\n');
out.push_str(SIGNAL_OUTBOX_DDL);
out.push('\n');
out.push_str(MIGRATION_TRACKING_DDL);
out.push('\n');
out.push_str(CACHE_INVALIDATE_DDL);
out.push('\n');
out.push_str(concat!(
"CREATE OR REPLACE FUNCTION _pylon.array_subscript(arr anyarray, idx bigint)\n",
"\tRETURNS anyelement\n",
"\tLANGUAGE plpgsql STABLE PARALLEL SAFE\n",
"AS $$\n",
"DECLARE\n",
" element_index bigint := CASE WHEN idx < 0 THEN idx + cardinality(arr) ELSE idx END;\n",
"BEGIN\n",
" IF element_index < 0 OR element_index >= cardinality(arr) THEN\n",
" RAISE EXCEPTION 'array index % is out of bounds', idx\n",
" USING ERRCODE = 'array_subscript_error';\n",
" END IF;\n",
" RETURN arr[element_index + 1];\n",
"END\n",
"$$;\n\n",
"CREATE OR REPLACE FUNCTION _pylon.raise_invalid_parameter(msg text)\n",
"\tRETURNS text\n",
"\tLANGUAGE plpgsql IMMUTABLE PARALLEL SAFE STRICT\n",
"AS $$\n",
"BEGIN\n",
" RAISE EXCEPTION '%', msg USING ERRCODE = 'invalid_parameter_value';\n",
"END\n",
"$$;\n\n",
"CREATE OR REPLACE FUNCTION _pylon.duration_in(val text)\n",
"\tRETURNS interval\n",
"\tLANGUAGE plpgsql IMMUTABLE PARALLEL SAFE STRICT\n",
"AS $$\n",
"DECLARE\n",
" parsed interval := val::interval;\n",
"BEGIN\n",
" IF date_part('year', parsed) != 0 OR date_part('month', parsed) != 0\n",
" OR date_part('day', parsed) != 0 THEN\n",
" RAISE EXCEPTION 'invalid input syntax for type duration: %', quote_literal(val)\n",
" USING ERRCODE = 'invalid_datetime_format',\n",
" HINT = 'Day, month and year units cannot be used for duration.';\n",
" END IF;\n",
" RETURN parsed;\n",
"END\n",
"$$;\n\n",
"CREATE OR REPLACE FUNCTION _pylon.date_duration_in(val text)\n",
"\tRETURNS interval\n",
"\tLANGUAGE plpgsql IMMUTABLE PARALLEL SAFE STRICT\n",
"AS $$\n",
"DECLARE\n",
" parsed interval := val::interval;\n",
"BEGIN\n",
" IF date_part('epoch', parsed - date_trunc('day', parsed)) != 0 THEN\n",
" RAISE EXCEPTION 'invalid input syntax for type cal::date_duration: %', quote_literal(val)\n",
" USING ERRCODE = 'invalid_datetime_format',\n",
" HINT = 'Units smaller than days cannot be used for cal::date_duration.';\n",
" END IF;\n",
" RETURN parsed;\n",
"END\n",
"$$;\n\n",
"CREATE OR REPLACE FUNCTION _pylon.str_subscript(s text, idx bigint)\n",
"\tRETURNS text\n",
"\tLANGUAGE plpgsql STABLE PARALLEL SAFE\n",
"AS $$\n",
"DECLARE\n",
" element_index bigint := CASE WHEN idx < 0 THEN idx + char_length(s) ELSE idx END;\n",
"BEGIN\n",
" IF element_index < 0 OR element_index >= char_length(s) THEN\n",
" RAISE EXCEPTION 'string index % is out of bounds', idx\n",
" USING ERRCODE = 'array_subscript_error';\n",
" END IF;\n",
" RETURN substr(s, (element_index + 1)::int, 1);\n",
"END\n",
"$$;\n\n",
"CREATE OR REPLACE FUNCTION _pylon.str_subscript(s bytea, idx bigint)\n",
"\tRETURNS bytea\n",
"\tLANGUAGE plpgsql STABLE PARALLEL SAFE\n",
"AS $$\n",
"DECLARE\n",
" element_index bigint := CASE WHEN idx < 0 THEN idx + length(s) ELSE idx END;\n",
"BEGIN\n",
" IF element_index < 0 OR element_index >= length(s) THEN\n",
" RAISE EXCEPTION 'byte string index % is out of bounds', idx\n",
" USING ERRCODE = 'array_subscript_error';\n",
" END IF;\n",
" RETURN substr(s, (element_index + 1)::int, 1);\n",
"END\n",
"$$;\n\n",
));
for desc in super::registry() {
if let ImplStrategy::PylonFunction(def) = &desc.impl_strategy {
out.push_str(&render_function(desc, def));
out.push('\n');
}
}
out
}
#[cfg(test)]
mod tests {
use super::{INTERNAL_SCHEMA_VERSION, export_stdlib};
#[test]
fn ddl_smoke() {
let ddl = export_stdlib();
let fn_count = ddl.matches("CREATE OR REPLACE FUNCTION").count();
assert!(ddl.starts_with("CREATE SCHEMA IF NOT EXISTS _pylon;"));
assert!(fn_count > 0, "no functions generated");
assert!(ddl.contains("_pylon.to_bool"), "to_bool missing");
assert!(ddl.contains("_pylon.enumerate"), "enumerate missing");
assert!(ddl.contains("_pylon.datetime_get"), "datetime_get missing");
assert!(
!ddl.contains("_pylon.range("),
"range must not be installed (TranspilerIntrinsic)"
);
assert!(!ddl.contains("_pylon.multirange("), "multirange must not be installed");
eprintln!("export_stdlib: {} PylonFunction overloads installed", fn_count);
}
#[test]
fn ddl_to_bool_has_three_overloads() {
let ddl = export_stdlib();
let count = ddl.matches("_pylon.to_bool(").count();
assert_eq!(count, 3, "expected int2/int4/int8 overloads; got {count}");
}
#[test]
fn ddl_enumerate_returns_table() {
let ddl = export_stdlib();
assert!(ddl.contains("RETURNS TABLE(index bigint, value anyelement)"));
}
#[test]
fn ddl_json_get_uses_variadic() {
let ddl = export_stdlib();
assert!(ddl.contains("VARIADIC path text[]"), "json_get must use VARIADIC");
}
#[test]
fn ddl_installs_cache_invalidate_notify_function() {
let ddl = export_stdlib();
assert!(ddl.contains("CREATE OR REPLACE FUNCTION _pylon.notify_cache_invalidate()"));
assert!(ddl.contains("pg_notify('pylon_cache_invalidate', TG_TABLE_SCHEMA || '.' || TG_TABLE_NAME)"));
}
#[test]
fn internal_schema_version_literal_matches_the_constant() {
assert_eq!(internal_schema_version_literal!(), INTERNAL_SCHEMA_VERSION.to_string(),);
}
#[test]
fn the_internal_version_marker_is_created_and_upserted() {
let ddl = export_stdlib();
assert!(ddl.contains("CREATE TABLE IF NOT EXISTS _pylon.\"Internal\""));
assert!(
ddl.contains("ON CONFLICT (singleton) DO UPDATE SET version = 1"),
"got:\n{ddl}"
);
}
}