AiDOPWarehouse/doc/db/mdp/02_mdp_pipeline_orchestration.sql

-- 数据中台：管道编排表（方案 §4.4.3）
-- 引擎未上线前可仅建表占位；调度仍可用独立 IJob（如 S3 S3MdpSyncTransformJob）。

SET NAMES utf8mb4;

CREATE TABLE IF NOT EXISTS mdp_pipeline (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    tenant_id BIGINT NOT NULL,
    pipeline_code VARCHAR(50) NOT NULL COMMENT 'FULL_SYNC / INCR_SYNC / S4_PO_SYNC',
    pipeline_name VARCHAR(200) NOT NULL,
    pipeline_type VARCHAR(20) DEFAULT 'SCHEDULED' COMMENT 'SCHEDULED / MANUAL / EVENT',
    cron_expression VARCHAR(50),
    max_concurrent_stage INT DEFAULT 2,
    timeout_minutes INT DEFAULT 30,
    retry_on_fail TINYINT DEFAULT 1,
    max_retries INT DEFAULT 2,
    is_enabled TINYINT DEFAULT 1,
    description TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME,
    UNIQUE KEY uk_pipeline (tenant_id, pipeline_code)
) COMMENT='数据同步管道定义';

CREATE TABLE IF NOT EXISTS mdp_pipeline_stage (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    tenant_id BIGINT NOT NULL,
    pipeline_code VARCHAR(50) NOT NULL,
    stage_code VARCHAR(50) NOT NULL,
    stage_name VARCHAR(200) NOT NULL,
    seq_no INT NOT NULL,
    depends_on_stage VARCHAR(256),
    parallel_within TINYINT DEFAULT 1,
    on_stage_fail VARCHAR(20) DEFAULT 'STOP' COMMENT 'STOP / CONTINUE / SKIP_STAGE',
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_stage (tenant_id, pipeline_code, stage_code),
    INDEX idx_pipeline (tenant_id, pipeline_code, seq_no)
) COMMENT='管道阶段定义';

CREATE TABLE IF NOT EXISTS mdp_pipeline_exec (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    tenant_id BIGINT NOT NULL,
    pipeline_code VARCHAR(50) NOT NULL,
    exec_no VARCHAR(64) NOT NULL,
    trigger_type VARCHAR(20) NOT NULL,
    status VARCHAR(20) DEFAULT 'PENDING',
    current_stage VARCHAR(50),
    started_at DATETIME,
    finished_at DATETIME,
    duration_seconds INT,
    error_msg TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_exec_no (tenant_id, exec_no),
    INDEX idx_pipeline (tenant_id, pipeline_code, created_at),
    INDEX idx_status (tenant_id, status)
) COMMENT='管道执行实例';

CREATE TABLE IF NOT EXISTS mdp_stage_exec (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    tenant_id BIGINT NOT NULL,
    exec_no VARCHAR(64) NOT NULL,
    stage_code VARCHAR(50) NOT NULL,
    seq_no INT NOT NULL,
    status VARCHAR(20) DEFAULT 'PENDING',
    total_tasks INT DEFAULT 0,
    success_tasks INT DEFAULT 0,
    failed_tasks INT DEFAULT 0,
    started_at DATETIME,
    finished_at DATETIME,
    duration_seconds INT,
    error_msg TEXT,
    INDEX idx_exec (tenant_id, exec_no),
    INDEX idx_stage (tenant_id, stage_code)
) COMMENT='阶段执行实例';