-- Generated by log-model/scripts/generate_sdm_event_ddl.py. Do not edit manually.
-- Source: log-model/contracts/hybrid-event/projection-registry.v1.json (hybrid-v1).
-- Production table name is sdm_event; registry target_table sdm_event_target is the shadow name.
-- Kafka hybrid-v1 -> sdm_event。jsonpaths 使用 Kafka 物理字段名（roles_obj 等）。
-- Placeholders: __DORIS_DB__ __KAFKA_BROKERS__ __KAFKA_TOPIC__ __KAFKA_OFFSETS__
-- 禁止写入兼容别名。首次生产用 OFFSET_END。
USE __DORIS_DB__;

CREATE ROUTINE LOAD sdm_events_load_hybrid_v1
ON sdm_event
COLUMNS(
    tenant_id_raw,
    occur_time_ms,
    event_id_raw,
    ingest_time_ms,
    parse_time_ms,
    schema_version,
    mapping_id,
    data_src_vendor,
    data_src_product,
    data_src_category,
    data_src_instance_id,
    log_type,
    log_level,
    log_name,
    log_id,
    record_kind,
    event_domain,
    event_type,
    operation,
    outcome,
    severity,
    log_content,
    source_ip,
    source_mac,
    source_port,
    source_user,
    source_account,
    source_host,
    source_process_name,
    carrier_process_name,
    carrier_process_guid,
    carrier_process_pid,
    target_ip,
    target_mac,
    target_port,
    target_user,
    target_host,
    target_service,
    target_domain,
    target_url,
    target_file_path,
    target_file_name,
    target_file_sha256,
    observer_product,
    observer_type,
    observer_vendor,
    device_ip,
    network_protocol,
    network_direction,
    network_session_id,
    application_name,
    http_method,
    http_status,
    http_host,
    k8s_namespace,
    k8s_pod_name,
    source_finding_title,
    source_finding_severity,
    source_finding_category,
    source_finding_signature_id,
    source_finding_action,
    source_finding_original_id,
    roles_obj,
    facets_obj,
    source_finding_obj,
    extensions_obj,
    tenant_id = TRIM(tenant_id_raw),
    occur_time = FROM_UNIXTIME(IF(CAST(occur_time_ms AS DOUBLE) >= 100000000000, CAST(occur_time_ms AS DOUBLE) / 1000, CAST(occur_time_ms AS DOUBLE))),
    event_id = TRIM(event_id_raw),
    ingest_time = IF(ingest_time_ms IS NULL, NOW(3), FROM_UNIXTIME(IF(CAST(ingest_time_ms AS DOUBLE) >= 100000000000, CAST(ingest_time_ms AS DOUBLE) / 1000, CAST(ingest_time_ms AS DOUBLE)))),
    parse_time = IF(parse_time_ms IS NULL, NOW(3), FROM_UNIXTIME(IF(CAST(parse_time_ms AS DOUBLE) >= 100000000000, CAST(parse_time_ms AS DOUBLE) / 1000, CAST(parse_time_ms AS DOUBLE))))
)
PROPERTIES (
    "format" = "json",
    "jsonpaths" = '["$.tenant_id","$.occur_time","$.event_id","$.ingest_time","$.parse_time","$.schema_version","$.mapping_id","$.data_src_vendor","$.data_src_product","$.data_src_category","$.data_src_instance_id","$.log_type","$.log_level","$.log_name","$.log_id","$.record_kind","$.event_domain","$.event_type","$.operation","$.outcome","$.severity","$.log_content","$.source_ip","$.source_mac","$.source_port","$.source_user","$.source_account","$.source_host","$.source_process_name","$.carrier_process_name","$.carrier_process_guid","$.carrier_process_pid","$.target_ip","$.target_mac","$.target_port","$.target_user","$.target_host","$.target_service","$.target_domain","$.target_url","$.target_file_path","$.target_file_name","$.target_file_sha256","$.observer_product","$.observer_type","$.observer_vendor","$.device_ip","$.network_protocol","$.network_direction","$.network_session_id","$.application_name","$.http_method","$.http_status","$.http_host","$.k8s_namespace","$.k8s_pod_name","$.source_finding_title","$.source_finding_severity","$.source_finding_category","$.source_finding_signature_id","$.source_finding_action","$.source_finding_original_id","$.roles_obj","$.facets_obj","$.source_finding_obj","$.extensions_obj"]',
    "strip_outer_array" = "false",
    "max_filter_ratio" = "1",
    "max_error_number" = "1000000000",
    "timezone" = "Etc/UTC"
)
FROM KAFKA (
    "kafka_broker_list" = "__KAFKA_BROKERS__",
    "kafka_topic" = "__KAFKA_TOPIC__",
    "kafka_partitions" = "__KAFKA_PARTITIONS__",
    "kafka_offsets" = "__KAFKA_OFFSETS__"
);
