Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

backend: finalize v2 migration (v2 phase 4 - final) #5155

Merged
merged 3 commits into from
Feb 13, 2025
Merged

backend: finalize v2 migration (v2 phase 4 - final) #5155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6a2a04a
backend: finalize v2 migration (v2 phase 4 - final)
uael Jan 20, 2025
bebae99
fix conflicts
rubenfiszel Feb 13, 2025
ae9edec
update migration
rubenfiszel Feb 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
251 changes: 71 additions & 180 deletions backend/tests/fixtures/base.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,183 +53,74 @@ EXECUTE FUNCTION "notify_queue" ();
WHEN (NEW.flow_status IS DISTINCT FROM OLD.flow_status)
EXECUTE FUNCTION "notify_queue" ();

-- TODO(uael): remove before phase 4
CREATE OR REPLACE FUNCTION zzz_v2_job_queue_integrity_check() RETURNS TRIGGER AS $$
DECLARE job v2_job;
DECLARE job_runtime v2_job_runtime;
DECLARE job_status v2_job_status;
BEGIN
IF (OLD.canceled_by IS NOT NULL) IS DISTINCT FROM OLD.__canceled THEN
RAISE EXCEPTION 'canceled mismatch';
END IF;
-- v2_job:
SELECT * INTO job FROM v2_job WHERE id = OLD.id;
IF job.tag IS DISTINCT FROM OLD.tag THEN
RAISE EXCEPTION 'tag mismatch';
END IF;
IF job.workspace_id IS DISTINCT FROM OLD.workspace_id THEN
RAISE EXCEPTION 'workspace_id mismatch';
END IF;
IF job.created_at IS DISTINCT FROM OLD.created_at THEN
RAISE EXCEPTION 'created_at mismatch';
END IF;
IF job.created_by IS DISTINCT FROM OLD.__created_by THEN
RAISE EXCEPTION 'created_by mismatch';
END IF;
IF job.permissioned_as IS DISTINCT FROM OLD.__permissioned_as THEN
RAISE EXCEPTION 'permissioned_as mismatch';
END IF;
IF job.permissioned_as_email IS DISTINCT FROM OLD.__email THEN
RAISE EXCEPTION 'permissioned_as_email mismatch';
END IF;
IF job.kind IS DISTINCT FROM OLD.__job_kind THEN
RAISE EXCEPTION 'kind mismatch';
END IF;
IF job.runnable_id IS DISTINCT FROM OLD.__script_hash THEN
RAISE EXCEPTION 'runnable_id mismatch';
END IF;
IF job.runnable_path IS DISTINCT FROM OLD.__script_path THEN
RAISE EXCEPTION 'runnable_path mismatch';
END IF;
IF job.parent_job IS DISTINCT FROM OLD.__parent_job THEN
RAISE EXCEPTION 'parent_job mismatch';
END IF;
IF job.script_lang IS DISTINCT FROM OLD.__language THEN
RAISE EXCEPTION 'script_lang mismatch';
END IF;
IF job.script_entrypoint_override IS DISTINCT FROM NULLIF(OLD.__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR')
AND OLD.__args->>'reason' IS DISTINCT FROM 'PREPROCESSOR_ARGS_ARE_DISCARDED'
THEN
RAISE EXCEPTION 'script_entrypoint_override mismatch';
END IF;
IF job.flow_step_id IS DISTINCT FROM OLD.__flow_step_id THEN
RAISE EXCEPTION 'flow_step_id mismatch';
END IF;
IF (job.flow_step_id IS NOT NULL) IS DISTINCT FROM OLD.__is_flow_step THEN
RAISE EXCEPTION 'is_flow_step mismatch';
END IF;
IF job.flow_innermost_root_job IS DISTINCT FROM OLD.__root_job THEN
RAISE EXCEPTION 'flow_innermost_root_job mismatch';
END IF;
IF job.trigger IS DISTINCT FROM OLD.__schedule_path THEN
RAISE EXCEPTION 'trigger mismatch';
END IF;
IF job.same_worker IS DISTINCT FROM OLD.__same_worker THEN
RAISE EXCEPTION 'same_worker mismatch';
END IF;
IF job.visible_to_owner IS DISTINCT FROM OLD.__visible_to_owner THEN
RAISE EXCEPTION 'visible_to_owner mismatch';
END IF;
IF job.concurrent_limit IS DISTINCT FROM OLD.__concurrent_limit THEN
RAISE EXCEPTION 'concurrent_limit mismatch';
END IF;
IF job.concurrency_time_window_s IS DISTINCT FROM OLD.__concurrency_time_window_s THEN
RAISE EXCEPTION 'concurrency_time_window_s mismatch';
END IF;
IF job.cache_ttl IS DISTINCT FROM OLD.__cache_ttl THEN
RAISE EXCEPTION 'cache_ttl mismatch';
END IF;
IF job.timeout IS DISTINCT FROM OLD.__timeout THEN
RAISE EXCEPTION 'timeout mismatch';
END IF;
IF job.priority IS DISTINCT FROM OLD.priority THEN
RAISE EXCEPTION 'priority mismatch';
END IF;
IF job.args::TEXT IS DISTINCT FROM OLD.__args::TEXT AND OLD.__args->>'_ENTRYPOINT_OVERRIDE' IS DISTINCT FROM '__WM_PREPROCESSOR' THEN
RAISE EXCEPTION 'args mismatch';
END IF;
IF job.pre_run_error IS DISTINCT FROM OLD.__pre_run_error THEN
RAISE EXCEPTION 'pre_run_error mismatch';
END IF;
-- v2_job_runtime:
SELECT * INTO job_runtime FROM v2_job_runtime WHERE id = OLD.id;
IF job_runtime.ping IS DISTINCT FROM OLD.__last_ping THEN
RAISE EXCEPTION 'ping mismatch';
END IF;
IF job_runtime.memory_peak IS DISTINCT FROM OLD.__mem_peak THEN
RAISE EXCEPTION 'memory_peak mismatch';
END IF;
-- v2_job_status:
IF EXISTS(SELECT 1 FROM v2_job_status WHERE id = OLD.id) THEN
SELECT * INTO job_status FROM v2_job_status WHERE id = OLD.id;
IF COALESCE(job_status.flow_status, job_status.workflow_as_code_status)::TEXT IS DISTINCT FROM OLD.__flow_status::TEXT
THEN
RAISE EXCEPTION 'flow_status mismatch';
END IF;
IF job_status.flow_leaf_jobs::TEXT IS DISTINCT FROM OLD.__leaf_jobs::TEXT THEN
RAISE EXCEPTION 'leaf_jobs mismatch';
END IF;
END IF;
RETURN OLD;
END $$ LANGUAGE PLPGSQL;

CREATE OR REPLACE TRIGGER zzz_v2_job_queue_integrity_check_before_delete
BEFORE DELETE ON v2_job_queue
FOR EACH ROW
EXECUTE FUNCTION zzz_v2_job_queue_integrity_check();

-- TODO(uael): remove before phase 4
CREATE OR REPLACE FUNCTION zzz_v2_job_completed_integrity_check() RETURNS TRIGGER AS $$
DECLARE job v2_job;
BEGIN
IF (NEW.canceled_by IS NOT NULL) IS DISTINCT FROM NEW.__canceled THEN
RAISE EXCEPTION 'canceled mismatch';
END IF;
SELECT * INTO job FROM v2_job WHERE id = NEW.id;
IF job.tag IS DISTINCT FROM NEW.__tag THEN
RAISE EXCEPTION 'tag mismatch % %', job.tag, NEW.__tag;
END IF;
IF job.workspace_id IS DISTINCT FROM NEW.workspace_id THEN
RAISE EXCEPTION 'workspace_id mismatch';
END IF;
IF job.created_at IS DISTINCT FROM NEW.__created_at THEN
RAISE EXCEPTION 'created_at mismatch';
END IF;
IF job.created_by IS DISTINCT FROM NEW.__created_by THEN
RAISE EXCEPTION 'created_by mismatch';
END IF;
IF job.permissioned_as IS DISTINCT FROM NEW.__permissioned_as THEN
RAISE EXCEPTION 'permissioned_as mismatch';
END IF;
IF job.permissioned_as_email IS DISTINCT FROM NEW.__email THEN
RAISE EXCEPTION 'permissioned_as_email mismatch';
END IF;
IF job.kind IS DISTINCT FROM NEW.__job_kind THEN
RAISE EXCEPTION 'kind mismatch';
END IF;
IF job.runnable_id IS DISTINCT FROM NEW.__script_hash THEN
RAISE EXCEPTION 'runnable_id mismatch';
END IF;
IF job.runnable_path IS DISTINCT FROM NEW.__script_path THEN
RAISE EXCEPTION 'runnable_path mismatch';
END IF;
IF job.parent_job IS DISTINCT FROM NEW.__parent_job THEN
RAISE EXCEPTION 'parent_job mismatch';
END IF;
IF job.script_lang IS DISTINCT FROM NEW.__language THEN
RAISE EXCEPTION 'script_lang mismatch';
END IF;
IF job.script_entrypoint_override IS DISTINCT FROM NULLIF(NEW.__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR')
AND NEW.__args->>'reason' IS DISTINCT FROM 'PREPROCESSOR_ARGS_ARE_DISCARDED'
THEN
RAISE EXCEPTION 'script_entrypoint_override mismatch';
END IF;
IF (job.flow_step_id IS NOT NULL) IS DISTINCT FROM NEW.__is_flow_step THEN
RAISE EXCEPTION 'is_flow_step mismatch';
END IF;
IF job.trigger IS DISTINCT FROM NEW.__schedule_path THEN
RAISE EXCEPTION 'trigger mismatch';
END IF;
IF job.visible_to_owner IS DISTINCT FROM NEW.__visible_to_owner THEN
RAISE EXCEPTION 'visible_to_owner mismatch';
END IF;
IF job.args::TEXT IS DISTINCT FROM NEW.__args::TEXT AND NEW.__args->>'_ENTRYPOINT_OVERRIDE' IS DISTINCT FROM '__WM_PREPROCESSOR' THEN
RAISE EXCEPTION 'args mismatch';
END IF;
RETURN NEW;
END $$ LANGUAGE PLPGSQL;

CREATE OR REPLACE TRIGGER zzz_v2_job_completed_integrity_check_after_insert
AFTER INSERT ON v2_job_completed
FOR EACH ROW
EXECUTE FUNCTION zzz_v2_job_completed_integrity_check();
-- Apply phase 4:
DROP FUNCTION IF EXISTS v2_job_after_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_completed_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_completed_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_after_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_status_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_status_before_update CASCADE;

DROP VIEW IF EXISTS completed_job, completed_job_view, job, queue, queue_view CASCADE;

ALTER TABLE v2_job_queue
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __last_ping CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __flow_status CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __same_worker CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __pre_run_error CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __mem_peak CASCADE,
DROP COLUMN IF EXISTS __root_job CASCADE,
DROP COLUMN IF EXISTS __leaf_jobs CASCADE,
DROP COLUMN IF EXISTS __concurrent_limit CASCADE,
DROP COLUMN IF EXISTS __concurrency_time_window_s CASCADE,
DROP COLUMN IF EXISTS __timeout CASCADE,
DROP COLUMN IF EXISTS __flow_step_id CASCADE,
DROP COLUMN IF EXISTS __cache_ttl CASCADE;

LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
ALTER TABLE v2_job_completed
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __created_at CASCADE,
DROP COLUMN IF EXISTS __success CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __is_skipped CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __tag CASCADE,
DROP COLUMN IF EXISTS __priority CASCADE;
6 changes: 3 additions & 3 deletions backend/tests/fixtures/result_format.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@ INSERT INTO public.v2_job (
'1eecb96a-c8b0-4a3d-b1b6-087878c55e41', 'test-workspace', 'test-user', '2023-01-01 00:00:00', 'script', 'postgresql'
);

INSERT INTO public.completed_job (
id, workspace_id, created_by, created_at, duration_ms, success, flow_status, result, job_kind, language
INSERT INTO public.v2_job_completed (
id, workspace_id, duration_ms, status, result_columns, result
) VALUES (
'1eecb96a-c8b0-4a3d-b1b6-087878c55e41', 'test-workspace', 'test-user', '2023-01-01 00:00:00', 1000, true, '{"_metadata": {"column_order": ["b", "a"]}}', '[{"a": "second", "b": "first"}]', 'script', 'postgresql'
'1eecb96a-c8b0-4a3d-b1b6-087878c55e41', 'test-workspace', 1000, 'success'::job_status, '{b,a}', '[{"a": "second", "b": "first"}]'
)
5 changes: 0 additions & 5 deletions backend/tests/worker.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4005,11 +4005,6 @@ mod job_payload {
let args = job.args.as_ref().unwrap();
assert_eq!(args.get("foo"), Some(&json!("bar")));
assert_eq!(args.get("bar"), Some(&json!("baz")));
// TODO: remove this check on v2 phase 4
assert_eq!(
job.flow_status.as_ref().unwrap().get("_metadata"),
Some(&json!({"preprocessed_args": true}))
);
assert_eq!(job.json_result().unwrap(), json!("Hello bar baz"));
let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id)
.fetch_one(db)
Expand Down
Loading