344 lines
16 KiB
PL/PgSQL
344 lines
16 KiB
PL/PgSQL
-- Data processing migration.
|
||
--
|
||
-- IMPORTANT: This file is intentionally NOT wired into application startup.
|
||
-- Apply it explicitly in a controlled deployment, or call
|
||
-- DataProcessStore.ensure_schema() from an administrative command.
|
||
|
||
BEGIN;
|
||
|
||
-- This migration targets the current runtime schema created by
|
||
-- 001_platform_runtime.sql. Refuse the UUID/JSONB target-design schema instead
|
||
-- of partially altering it with incompatible TEXT foreign keys.
|
||
DO $$
|
||
DECLARE
|
||
datasets_id_type TEXT;
|
||
BEGIN
|
||
SELECT format_type(a.atttypid, a.atttypmod)
|
||
INTO datasets_id_type
|
||
FROM pg_attribute a
|
||
JOIN pg_class c ON c.oid = a.attrelid
|
||
JOIN pg_namespace n ON n.oid = c.relnamespace
|
||
WHERE n.nspname = current_schema()
|
||
AND c.relname = 'datasets'
|
||
AND a.attname = 'id'
|
||
AND a.attnum > 0
|
||
AND NOT a.attisdropped;
|
||
IF datasets_id_type IS NULL THEN
|
||
RAISE EXCEPTION '002_data_process.sql requires 001_platform_runtime.sql first';
|
||
END IF;
|
||
IF datasets_id_type <> 'text' THEN
|
||
RAISE EXCEPTION
|
||
'002_data_process.sql supports only the current TEXT runtime schema; found datasets.id type %',
|
||
datasets_id_type;
|
||
END IF;
|
||
END $$;
|
||
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS source_task_id TEXT;
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS size_bytes BIGINT NOT NULL DEFAULT 0;
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS record_count BIGINT NOT NULL DEFAULT 0;
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAULT '{}';
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS tenant_id TEXT;
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS project_id TEXT;
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS owner_id TEXT;
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS created_by TEXT;
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT now();
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now();
|
||
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
|
||
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS storage_object_id TEXT;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS current_version_id TEXT;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS size_bytes BIGINT NOT NULL DEFAULT 0;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS record_count BIGINT NOT NULL DEFAULT 0;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS file_format VARCHAR(40);
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS checksum_sha256 CHAR(64);
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS version_no INTEGER NOT NULL DEFAULT 1;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS source_task_id TEXT;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS tenant_id TEXT;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS project_id TEXT;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS created_by TEXT;
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAULT '{}';
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT now();
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now();
|
||
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
|
||
|
||
CREATE TABLE IF NOT EXISTS data_process_tasks (
|
||
id TEXT PRIMARY KEY,
|
||
name VARCHAR(150) NOT NULL,
|
||
description TEXT,
|
||
status VARCHAR(20) NOT NULL DEFAULT 'pending'
|
||
CHECK (status IN ('pending', 'running', 'completed', 'failed', 'stopped')),
|
||
process_type VARCHAR(20) NOT NULL
|
||
CHECK (process_type IN ('structured', 'unstructured', 'external')),
|
||
source_dataset_id TEXT REFERENCES datasets(id) ON DELETE SET NULL,
|
||
output_dataset_id TEXT REFERENCES datasets(id) ON DELETE SET NULL,
|
||
config TEXT NOT NULL DEFAULT '{}',
|
||
progress NUMERIC(5,2) NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100),
|
||
input_count BIGINT NOT NULL DEFAULT 0 CHECK (input_count >= 0),
|
||
output_count BIGINT NOT NULL DEFAULT 0 CHECK (output_count >= 0),
|
||
filtered_count BIGINT NOT NULL DEFAULT 0 CHECK (filtered_count >= 0),
|
||
duplicate_count BIGINT NOT NULL DEFAULT 0 CHECK (duplicate_count >= 0),
|
||
error_count BIGINT NOT NULL DEFAULT 0 CHECK (error_count >= 0),
|
||
failure_reason TEXT,
|
||
generation_run_id TEXT,
|
||
results_confirmed BOOLEAN NOT NULL DEFAULT TRUE,
|
||
workflow_step VARCHAR(20) NOT NULL DEFAULT 'create'
|
||
CHECK (workflow_step IN ('create', 'model', 'upload', 'preview', 'generate', 'results')),
|
||
preview_status VARCHAR(20) NOT NULL DEFAULT 'idle'
|
||
CHECK (preview_status IN ('idle', 'queued', 'running', 'completed', 'failed', 'cancelled')),
|
||
preview_progress NUMERIC(5,2) NOT NULL DEFAULT 0
|
||
CHECK (preview_progress >= 0 AND preview_progress <= 100),
|
||
preview_run_id TEXT,
|
||
preview_failure_reason TEXT,
|
||
preview_total_files INTEGER NOT NULL DEFAULT 0 CHECK (preview_total_files >= 0),
|
||
preview_completed_files INTEGER NOT NULL DEFAULT 0 CHECK (preview_completed_files >= 0),
|
||
tenant_id TEXT,
|
||
project_id TEXT,
|
||
owner_id TEXT,
|
||
approval_status VARCHAR(30) NOT NULL DEFAULT 'not_required',
|
||
created_by TEXT,
|
||
updated_by TEXT,
|
||
deleted_by TEXT,
|
||
started_at TIMESTAMPTZ,
|
||
completed_at TIMESTAMPTZ,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
deleted_at TIMESTAMPTZ
|
||
);
|
||
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS generation_run_id TEXT;
|
||
-- 历史任务在引入六步确认流程前已经完成审核，默认保留为已确认；
|
||
-- 新任务由创建接口显式写入 FALSE，并在第六步确认后转为 TRUE。
|
||
ALTER TABLE data_process_tasks
|
||
ADD COLUMN IF NOT EXISTS results_confirmed BOOLEAN NOT NULL DEFAULT TRUE;
|
||
UPDATE data_process_tasks
|
||
SET results_confirmed=FALSE
|
||
WHERE status <> 'completed' AND results_confirmed=TRUE;
|
||
|
||
-- 先以可空列接入旧库，才能只回填历史行；随后再收紧默认值与约束。
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS workflow_step VARCHAR(20);
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_status VARCHAR(20);
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_progress NUMERIC(5,2);
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_run_id TEXT;
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_failure_reason TEXT;
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_total_files INTEGER;
|
||
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_completed_files INTEGER;
|
||
|
||
CREATE TEMP TABLE data_process_workflow_backfill_ids ON COMMIT DROP AS
|
||
SELECT id FROM data_process_tasks WHERE workflow_step IS NULL;
|
||
|
||
UPDATE data_process_tasks task
|
||
SET workflow_step = CASE
|
||
WHEN task.status IN ('running', 'failed', 'stopped') THEN 'generate'
|
||
WHEN task.status = 'completed' AND task.results_confirmed=FALSE THEN 'generate'
|
||
WHEN task.status = 'completed' THEN 'results'
|
||
ELSE 'create'
|
||
END
|
||
WHERE task.workflow_step IS NULL;
|
||
UPDATE data_process_tasks
|
||
SET preview_status='idle', preview_progress=0,
|
||
preview_total_files=0, preview_completed_files=0
|
||
WHERE preview_status IS NULL OR preview_progress IS NULL
|
||
OR preview_total_files IS NULL OR preview_completed_files IS NULL;
|
||
|
||
ALTER TABLE data_process_tasks ALTER COLUMN workflow_step SET DEFAULT 'create';
|
||
ALTER TABLE data_process_tasks ALTER COLUMN workflow_step SET NOT NULL;
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_status SET DEFAULT 'idle';
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_status SET NOT NULL;
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_progress SET DEFAULT 0;
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_progress SET NOT NULL;
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_total_files SET DEFAULT 0;
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_total_files SET NOT NULL;
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_completed_files SET DEFAULT 0;
|
||
ALTER TABLE data_process_tasks ALTER COLUMN preview_completed_files SET NOT NULL;
|
||
|
||
DO $$
|
||
BEGIN
|
||
IF NOT EXISTS (
|
||
SELECT 1 FROM pg_constraint
|
||
WHERE conrelid='data_process_tasks'::regclass
|
||
AND conname='ck_data_process_tasks_workflow_step'
|
||
) THEN
|
||
ALTER TABLE data_process_tasks ADD CONSTRAINT ck_data_process_tasks_workflow_step
|
||
CHECK (workflow_step IN ('create', 'model', 'upload', 'preview', 'generate', 'results'));
|
||
END IF;
|
||
IF NOT EXISTS (
|
||
SELECT 1 FROM pg_constraint
|
||
WHERE conrelid='data_process_tasks'::regclass
|
||
AND conname='ck_data_process_tasks_preview_status'
|
||
) THEN
|
||
ALTER TABLE data_process_tasks ADD CONSTRAINT ck_data_process_tasks_preview_status
|
||
CHECK (preview_status IN ('idle', 'queued', 'running', 'completed', 'failed', 'cancelled'));
|
||
END IF;
|
||
IF NOT EXISTS (
|
||
SELECT 1 FROM pg_constraint
|
||
WHERE conrelid='data_process_tasks'::regclass
|
||
AND conname='ck_data_process_tasks_preview_progress'
|
||
) THEN
|
||
ALTER TABLE data_process_tasks ADD CONSTRAINT ck_data_process_tasks_preview_progress
|
||
CHECK (preview_progress >= 0 AND preview_progress <= 100);
|
||
END IF;
|
||
IF NOT EXISTS (
|
||
SELECT 1 FROM pg_constraint
|
||
WHERE conrelid='data_process_tasks'::regclass
|
||
AND conname='ck_data_process_tasks_preview_file_counts'
|
||
) THEN
|
||
ALTER TABLE data_process_tasks ADD CONSTRAINT ck_data_process_tasks_preview_file_counts
|
||
CHECK (preview_total_files >= 0 AND preview_completed_files >= 0
|
||
AND preview_completed_files <= preview_total_files);
|
||
END IF;
|
||
END $$;
|
||
|
||
CREATE UNIQUE INDEX IF NOT EXISTS uq_data_process_tasks_name_alive
|
||
ON data_process_tasks(name) WHERE deleted_at IS NULL;
|
||
CREATE INDEX IF NOT EXISTS idx_data_process_tasks_scope_status
|
||
ON data_process_tasks(tenant_id, project_id, status, created_at DESC)
|
||
WHERE deleted_at IS NULL;
|
||
CREATE INDEX IF NOT EXISTS idx_data_process_tasks_creator_created
|
||
ON data_process_tasks(created_by, created_at DESC) WHERE deleted_at IS NULL;
|
||
|
||
CREATE TABLE IF NOT EXISTS data_process_source_files (
|
||
id TEXT PRIMARY KEY,
|
||
task_id TEXT NOT NULL REFERENCES data_process_tasks(id) ON DELETE CASCADE,
|
||
storage_object_id TEXT,
|
||
name TEXT NOT NULL,
|
||
size_bytes BIGINT NOT NULL DEFAULT 0 CHECK (size_bytes >= 0),
|
||
record_count BIGINT NOT NULL DEFAULT 0 CHECK (record_count >= 0),
|
||
file_format VARCHAR(40),
|
||
checksum_sha256 CHAR(64) NOT NULL,
|
||
version_no INTEGER NOT NULL DEFAULT 1 CHECK (version_no > 0),
|
||
content TEXT NOT NULL,
|
||
content_preview TEXT,
|
||
metadata TEXT NOT NULL DEFAULT '{}',
|
||
tenant_id TEXT,
|
||
project_id TEXT,
|
||
created_by TEXT,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
deleted_at TIMESTAMPTZ
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_data_process_source_files_task
|
||
ON data_process_source_files(task_id, created_at) WHERE deleted_at IS NULL;
|
||
CREATE UNIQUE INDEX IF NOT EXISTS uq_data_process_source_checksum_alive
|
||
ON data_process_source_files(task_id, checksum_sha256) WHERE deleted_at IS NULL;
|
||
|
||
CREATE TABLE IF NOT EXISTS data_process_preview_items (
|
||
id TEXT PRIMARY KEY,
|
||
task_id TEXT NOT NULL REFERENCES data_process_tasks(id) ON DELETE CASCADE,
|
||
source_file_id TEXT REFERENCES data_process_source_files(id) ON DELETE CASCADE,
|
||
original_content TEXT NOT NULL DEFAULT '',
|
||
edited_content TEXT NOT NULL DEFAULT '',
|
||
source_start INTEGER CHECK (source_start IS NULL OR source_start >= 0),
|
||
source_end INTEGER CHECK (source_end IS NULL OR source_end >= 0),
|
||
source_start_line INTEGER CHECK (source_start_line IS NULL OR source_start_line > 0),
|
||
source_end_line INTEGER CHECK (source_end_line IS NULL OR source_end_line > 0),
|
||
token_count INTEGER NOT NULL DEFAULT 0 CHECK (token_count >= 0),
|
||
status VARCHAR(20) NOT NULL DEFAULT 'original'
|
||
CHECK (status IN ('original', 'modified', 'manual', 'invalid')),
|
||
quality_score TEXT NOT NULL DEFAULT '{}',
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
CHECK (source_start IS NULL OR source_end IS NULL OR source_end >= source_start),
|
||
CHECK (source_start_line IS NULL OR source_end_line IS NULL OR source_end_line >= source_start_line)
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_data_process_preview_task_file
|
||
ON data_process_preview_items(task_id, source_file_id, created_at);
|
||
|
||
-- 子表在新库中到这里才存在；只修复本次新增 workflow_step 前的历史任务。
|
||
UPDATE data_process_tasks task
|
||
SET workflow_step = CASE
|
||
WHEN EXISTS (
|
||
SELECT 1 FROM data_process_preview_items preview
|
||
WHERE preview.task_id=task.id
|
||
) THEN 'preview'
|
||
WHEN EXISTS (
|
||
SELECT 1 FROM data_process_source_files source_file
|
||
WHERE source_file.task_id=task.id AND source_file.deleted_at IS NULL
|
||
) THEN 'upload'
|
||
ELSE task.workflow_step
|
||
END
|
||
WHERE task.id IN (SELECT id FROM data_process_workflow_backfill_ids)
|
||
AND task.status='pending';
|
||
|
||
CREATE TABLE IF NOT EXISTS data_process_results (
|
||
id TEXT PRIMARY KEY,
|
||
task_id TEXT NOT NULL REFERENCES data_process_tasks(id) ON DELETE CASCADE,
|
||
preview_item_id TEXT REFERENCES data_process_preview_items(id) ON DELETE SET NULL,
|
||
instruction TEXT NOT NULL,
|
||
input TEXT NOT NULL DEFAULT '',
|
||
output TEXT NOT NULL,
|
||
original_instruction TEXT,
|
||
original_input TEXT,
|
||
original_output TEXT,
|
||
status VARCHAR(20) NOT NULL DEFAULT 'valid'
|
||
CHECK (status IN ('valid', 'modified', 'invalid')),
|
||
error TEXT,
|
||
split VARCHAR(20) CHECK (split IS NULL OR split IN ('train', 'validation', 'test')),
|
||
quality_score TEXT NOT NULL DEFAULT '{}',
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_data_process_results_task_status
|
||
ON data_process_results(task_id, status, id);
|
||
CREATE INDEX IF NOT EXISTS idx_data_process_results_task_split
|
||
ON data_process_results(task_id, split);
|
||
|
||
CREATE TABLE IF NOT EXISTS dataset_file_versions (
|
||
id TEXT PRIMARY KEY,
|
||
dataset_file_id TEXT NOT NULL REFERENCES dataset_files(id) ON DELETE CASCADE,
|
||
version_no INTEGER NOT NULL CHECK (version_no > 0),
|
||
storage_object_id TEXT NOT NULL,
|
||
content_preview TEXT,
|
||
description TEXT,
|
||
base_version_id TEXT REFERENCES dataset_file_versions(id) ON DELETE SET NULL,
|
||
size_bytes BIGINT NOT NULL DEFAULT 0 CHECK (size_bytes >= 0),
|
||
record_count BIGINT NOT NULL DEFAULT 0 CHECK (record_count >= 0),
|
||
checksum_sha256 CHAR(64) NOT NULL,
|
||
source_task_id TEXT REFERENCES data_process_tasks(id) ON DELETE SET NULL,
|
||
metadata TEXT NOT NULL DEFAULT '{}',
|
||
created_by TEXT,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||
);
|
||
|
||
ALTER TABLE dataset_file_versions ADD COLUMN IF NOT EXISTS source_task_id TEXT;
|
||
ALTER TABLE dataset_file_versions ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAULT '{}';
|
||
CREATE UNIQUE INDEX IF NOT EXISTS uq_dataset_file_versions_no_002
|
||
ON dataset_file_versions(dataset_file_id, version_no);
|
||
CREATE INDEX IF NOT EXISTS idx_dataset_file_versions_source_task_002
|
||
ON dataset_file_versions(source_task_id) WHERE source_task_id IS NOT NULL;
|
||
|
||
CREATE TABLE IF NOT EXISTS dataset_records (
|
||
id TEXT PRIMARY KEY,
|
||
dataset_id TEXT NOT NULL REFERENCES datasets(id) ON DELETE CASCADE,
|
||
dataset_file_id TEXT REFERENCES dataset_files(id) ON DELETE CASCADE,
|
||
version_id TEXT REFERENCES dataset_file_versions(id) ON DELETE CASCADE,
|
||
line_no INTEGER,
|
||
split VARCHAR(20) CHECK (split IS NULL OR split IN ('train', 'validation', 'test')),
|
||
instruction TEXT,
|
||
input TEXT,
|
||
output TEXT,
|
||
raw TEXT NOT NULL DEFAULT '{}',
|
||
status VARCHAR(20) NOT NULL DEFAULT 'valid'
|
||
CHECK (status IN ('valid', 'modified', 'invalid')),
|
||
source_task_id TEXT REFERENCES data_process_tasks(id) ON DELETE SET NULL,
|
||
source_result_id TEXT REFERENCES data_process_results(id) ON DELETE SET NULL,
|
||
preview_item_id TEXT REFERENCES data_process_preview_items(id) ON DELETE SET NULL,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||
);
|
||
|
||
ALTER TABLE dataset_records ADD COLUMN IF NOT EXISTS source_task_id TEXT;
|
||
ALTER TABLE dataset_records ADD COLUMN IF NOT EXISTS source_result_id TEXT;
|
||
ALTER TABLE dataset_records ADD COLUMN IF NOT EXISTS preview_item_id TEXT;
|
||
CREATE INDEX IF NOT EXISTS idx_dataset_records_dataset_002
|
||
ON dataset_records(dataset_id, id);
|
||
CREATE INDEX IF NOT EXISTS idx_dataset_records_source_task_002
|
||
ON dataset_records(source_task_id, source_result_id);
|
||
CREATE INDEX IF NOT EXISTS idx_datasets_source_task_002
|
||
ON datasets(source_task_id) WHERE source_task_id IS NOT NULL;
|
||
CREATE INDEX IF NOT EXISTS idx_dataset_files_source_task_002
|
||
ON dataset_files(source_task_id) WHERE source_task_id IS NOT NULL;
|
||
|
||
COMMIT;
|