ai-security-xdr/haobang-security-dm/upgrade-doc/database/01-schema-upgrade.sql

-- haobang-security-dm：旧备份基线 -> 当前版本结构升级。
-- 前置：已完成备份，并停止所有目标表写入者。
-- 本文件为一次性 DDL。发生任一错误立即停止，禁止整文件盲目重跑。

-- 1. Retry/DLT 可检索索引和人工回放对象。
CREATE TABLE syslog_failure_record (
  id VARCHAR2(36) NOT NULL,
  event_id VARCHAR2(80) NOT NULL,
  device_id VARCHAR2(64),
  original_topic VARCHAR2(255) NOT NULL,
  original_partition INT NOT NULL,
  original_offset BIGINT NOT NULL,
  current_topic VARCHAR2(255),
  first_received_at TIMESTAMP,
  failed_stage VARCHAR2(64) NOT NULL,
  error_code VARCHAR2(128),
  error_type VARCHAR2(512),
  error_message VARCHAR2(2048),
  attempt_count INT DEFAULT 0 NOT NULL,
  next_retry_at TIMESTAMP,
  rule_id VARCHAR2(128),
  rule_version VARCHAR2(64),
  consumer_version VARCHAR2(64),
  point_time_ns BIGINT,
  original_payload CLOB NOT NULL,
  status VARCHAR2(32) DEFAULT 'OPEN' NOT NULL,
  replay_batch_id VARCHAR2(36),
  replay_count INT DEFAULT 0 NOT NULL,
  last_replay_at TIMESTAMP,
  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_syslog_failure_record PRIMARY KEY(id),
  CONSTRAINT uk_syslog_failure_origin
    UNIQUE(original_topic,original_partition,original_offset,failed_stage)
);

COMMENT ON TABLE syslog_failure_record IS 'Syslog Retry/DLT 异常消息可检索索引，Kafka DLT 为原始事实源';
COMMENT ON COLUMN syslog_failure_record.event_id IS '接入生成并在 Retry/Replay 中保持不变的事件标识';
COMMENT ON COLUMN syslog_failure_record.point_time_ns IS 'InfluxDB V2 Point 纳秒时间戳；旧 DLT 记录允许为空';
COMMENT ON COLUMN syslog_failure_record.original_payload IS '原始加密载荷，禁止接口明文返回';
COMMENT ON COLUMN syslog_failure_record.status IS 'OPEN/REPLAYED/RESOLVED 等补偿状态';

CREATE INDEX idx_syslog_failure_event
ON syslog_failure_record(event_id);

CREATE INDEX idx_syslog_failure_filter
ON syslog_failure_record(device_id,error_code,rule_version,first_received_at);

CREATE TABLE syslog_replay_batch (
  id VARCHAR2(36) NOT NULL,
  status VARCHAR2(32) NOT NULL,
  operator_name VARCHAR2(128) NOT NULL,
  reason VARCHAR2(1000) NOT NULL,
  rate_per_second INT NOT NULL,
  total_count INT NOT NULL,
  success_count INT DEFAULT 0 NOT NULL,
  failed_count INT DEFAULT 0 NOT NULL,
  duplicate_count INT DEFAULT 0 NOT NULL,
  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  completed_at TIMESTAMP,
  CONSTRAINT pk_syslog_replay_batch PRIMARY KEY(id)
);

COMMENT ON TABLE syslog_replay_batch IS '人工补偿回放批次及操作审计';

CREATE TABLE syslog_replay_item (
  id VARCHAR2(36) NOT NULL,
  replay_batch_id VARCHAR2(36) NOT NULL,
  failure_id VARCHAR2(36) NOT NULL,
  status VARCHAR2(32) NOT NULL,
  error_message VARCHAR2(1000),
  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_syslog_replay_item PRIMARY KEY(id),
  CONSTRAINT uk_syslog_replay_item UNIQUE(replay_batch_id,failure_id)
);

COMMENT ON TABLE syslog_replay_item IS '回放批次逐条状态';

CREATE INDEX idx_syslog_replay_pending
ON syslog_replay_item(replay_batch_id,status,created_at);

-- 2. 标准化、规则告警和算法告警幂等字段。
-- 只增加可空列；历史数据保持 NULL，不执行无法可靠还原版本/窗口的全表 datafix。
ALTER TABLE syslog_normal_data ADD rule_version VARCHAR2(64);
ALTER TABLE syslog_normal_alarm ADD rule_version VARCHAR2(64);

COMMENT ON COLUMN syslog_normal_data.rule_version IS '标准化结果使用的规则版本；确定性 id 的组成部分';
COMMENT ON COLUMN syslog_normal_alarm.rule_version IS '标准化结果使用的规则版本；确定性 id 的组成部分';

ALTER TABLE alarm ADD rule_id VARCHAR2(128);
ALTER TABLE alarm ADD rule_version VARCHAR2(64);
ALTER TABLE alarm ADD window_start TIMESTAMP;
ALTER TABLE alarm ADD group_key_hash VARCHAR2(64);

COMMENT ON COLUMN alarm.rule_id IS '生成告警的关联分析规则标识';
COMMENT ON COLUMN alarm.rule_version IS '生成告警的关联分析规则版本';
COMMENT ON COLUMN alarm.window_start IS '聚合告警业务窗口起点';
COMMENT ON COLUMN alarm.group_key_hash IS '规范化分组结果的稳定哈希';

ALTER TABLE alarm_visit ADD algorithm_id VARCHAR2(128);
ALTER TABLE alarm_visit ADD algorithm_version VARCHAR2(64);
ALTER TABLE alarm_visit ADD event_id VARCHAR2(80);
ALTER TABLE alarm_visit ADD result_type VARCHAR2(64);

COMMENT ON COLUMN alarm_visit.algorithm_id IS '生成结果的算法标识';
COMMENT ON COLUMN alarm_visit.algorithm_version IS '生成结果的算法版本';
COMMENT ON COLUMN alarm_visit.event_id IS '算法输入源事件的稳定 event_id';
COMMENT ON COLUMN alarm_visit.result_type IS '算法返回结果类型';

-- 3. 规则/算法窗口状态和多实例租约。
CREATE TABLE analysis_checkpoint (
  id VARCHAR2(36) NOT NULL,
  task_type VARCHAR2(32) NOT NULL,
  task_id VARCHAR2(128) NOT NULL,
  task_version VARCHAR2(64) NOT NULL,
  window_start TIMESTAMP NOT NULL,
  window_end TIMESTAMP NOT NULL,
  status VARCHAR2(32) NOT NULL,
  attempt_count INT DEFAULT 0 NOT NULL,
  input_count BIGINT DEFAULT 0 NOT NULL,
  output_count BIGINT DEFAULT 0 NOT NULL,
  error_code VARCHAR2(128),
  error_message VARCHAR2(2000),
  owner_id VARCHAR2(255),
  fencing_token BIGINT DEFAULT 0 NOT NULL,
  lease_until TIMESTAMP,
  started_at TIMESTAMP,
  completed_at TIMESTAMP,
  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_analysis_checkpoint PRIMARY KEY(id),
  CONSTRAINT uk_analysis_checkpoint_window
    UNIQUE(task_type,task_id,task_version,window_start,window_end),
  CONSTRAINT ck_analysis_checkpoint_status
    CHECK(status IN ('PENDING','RUNNING','COMPLETED','FAILED','RETRYING'))
);

COMMENT ON TABLE analysis_checkpoint IS '规则和算法独立窗口状态、连续成功水位与多实例租约';

CREATE INDEX idx_analysis_checkpoint_due
ON analysis_checkpoint(task_type,status,lease_until,window_end);

-- 4. 当前最终 DIRECT 统计正式表。
-- 旧备份基线没有中间版 change-log 对象，本升级直接建立最终小型聚合模型。
CREATE TABLE device_daily_stats (
  stat_date DATE NOT NULL,
  device_id BIGINT NOT NULL,
  parse_count BIGINT DEFAULT 0 NOT NULL,
  non_log_count BIGINT DEFAULT 0 NOT NULL,
  last_receive_time TIMESTAMP,
  projection_seq BIGINT DEFAULT 0 NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_device_daily_stats PRIMARY KEY(stat_date,device_id),
  CONSTRAINT ck_device_daily_stats_count CHECK(parse_count>=0 AND non_log_count>=0)
);

CREATE TABLE normalize_rule_daily_stats (
  stat_date DATE NOT NULL,
  normalize_rule_id BIGINT NOT NULL,
  max_log_time TIMESTAMP,
  projection_seq BIGINT DEFAULT 0 NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_normalize_rule_daily_stats PRIMARY KEY(stat_date,normalize_rule_id)
);

CREATE TABLE device_collect_stats (
  device_collect_id BIGINT NOT NULL,
  first_success_time TIMESTAMP,
  last_success_time TIMESTAMP,
  last_fail_time TIMESTAMP,
  projection_seq BIGINT DEFAULT 0 NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_device_collect_stats PRIMARY KEY(device_collect_id)
);

COMMENT ON TABLE device_daily_stats IS 'DIRECT 按业务日期和设备维护的物理行计数及入口时间';
COMMENT ON TABLE normalize_rule_daily_stats IS 'DIRECT 按业务日期和规则维护的最大日志时间';
COMMENT ON TABLE device_collect_stats IS 'DIRECT 维护采集任务首次成功及最后成功/失败时间';

-- 5. DIRECT 短期影子表和人工重建状态。
CREATE TABLE device_daily_stats_direct_shadow (
  stat_date DATE NOT NULL,
  device_id BIGINT NOT NULL,
  parse_count BIGINT DEFAULT 0 NOT NULL,
  non_log_count BIGINT DEFAULT 0 NOT NULL,
  last_receive_time TIMESTAMP,
  projection_seq BIGINT DEFAULT 0 NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_device_daily_direct_shadow PRIMARY KEY(stat_date,device_id),
  CONSTRAINT ck_device_daily_direct_count CHECK(parse_count>=0 AND non_log_count>=0)
);

CREATE TABLE normalize_rule_daily_stats_direct_shadow (
  stat_date DATE NOT NULL,
  normalize_rule_id BIGINT NOT NULL,
  max_log_time TIMESTAMP,
  projection_seq BIGINT DEFAULT 0 NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_rule_daily_direct_shadow PRIMARY KEY(stat_date,normalize_rule_id)
);

CREATE TABLE stats_direct_control (
  control_name VARCHAR2(64) NOT NULL,
  rebuild_required SMALLINT DEFAULT 0 NOT NULL,
  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
  CONSTRAINT pk_stats_direct_control PRIMARY KEY(control_name),
  CONSTRAINT ck_stats_direct_rebuild CHECK(rebuild_required IN (0,1))
);

INSERT INTO stats_direct_control(control_name,rebuild_required,updated_at)
VALUES ('SYSLOG_STATS',0,SYSDATE);

COMMENT ON TABLE device_daily_stats_direct_shadow IS 'DIRECT 短期对账设备影子表，正式 DIRECT 不读取';
COMMENT ON TABLE normalize_rule_daily_stats_direct_shadow IS 'DIRECT 短期对账规则影子表，正式 DIRECT 不读取';
COMMENT ON TABLE stats_direct_control IS 'DIRECT 人工降级和重建状态';

-- 6. 今日计数日切保护字段。
ALTER TABLE device_device ADD today_stats_date DATE;

COMMENT ON COLUMN device_device.today_stats_date IS 'today_parse_count/today_non_log_count 对应的上海业务日期';

COMMIT;