CREATE TABLE IF NOT EXISTS `abm-exporter-paimon`.`abm_exporter_db`.`ods_realtime_exporter_lakehouse` (
    `sls_time`  bigint,
    `cluster`  varchar,
    `content`  varchar
) PARTITIONED BY (cluster) with(
    'orc.write.batch-size' = '128',
    'file.format' = 'avro',
    'bucket' = '8'
);

insert into `abm-exporter-paimon`.`abm_exporter_db`.`ods_realtime_exporter_lakehouse` /*+ OPTIONS('orc.write-buffer-size' = '128','file.format' = 'avro') */
select
    `__timestamp__` as sls_time,
    `__topic__` as cluster,
    `content`
from source_k8s_meta;

CREATE TABLE IF NOT EXISTS `abm-exporter-paimon`.`abm_exporter_db`.`dwd_realtime_exporter_lakehouse` (
    type                 varchar,
    obj                 varchar,
    k8s_group           varchar,
    version             varchar,
    kind                varchar,
    namespace           varchar,
    name                varchar,
    resourceversion     varchar,
    region_id           varchar NOT NULL,
    cluster             varchar NOT NULL,
    watchname           varchar NOT NULL,
    watchsetuptimestamp bigint NOT NULL,
    uid                 varchar PRIMARY KEY NOT ENFORCED,
    collect_time        bigint NOT NULL,
    sls_time            bigint,
    cluster_id          varchar,
    region              varchar,
    zone                varchar
) with (
    'merge-engine' = 'partial-update',
    'changelog-producer' = 'lookup',
    'partial-update.ignore-delete'='true'
);

set 'table.exec.sink.upsert-materialize' = 'NONE';

insert into `abm-exporter-paimon`.`abm_exporter_db`.`dwd_realtime_exporter_lakehouse`  /*+ OPTIONS('orc.write-buffer-size' = '128','file.format' = 'avro') */
select
    json_value(`content`, '$.type')             as `type`,
    json_value(`content`, '$.obj')              as `obj`,
    json_value(`content`, '$.group')            as `k8s_group`,
    json_value(`content`, '$.version')          as `version`,
    json_value(`content`, '$.kind')             as `kind`,
    json_value(`content`, '$.namespace')        as `namespace`,
    json_value(`content`, '$.name')             as `name`,
    json_value(`content`, '$.resourceversion')  as `resourceversion`,
    'cn-zhangjiakou'                               as `region_id`,
    `cluster`                                as `cluster`,
    json_value(`content`, '$.watchname')        as `watchname`,
    cast(json_value(`content`, '$.watchsetuptimestamp') as bigint)    as `watchsetuptimestamp`,
    json_value(`content`, '$.uid')              as `uid`,
    cast(UNIX_TIMESTAMP() as bigint)                      as `collect_time`,
    `sls_time`                            as `sls_time`,
    NULL,
    NULL,
    NULL
from (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY `cluster`, json_value(`content`, '$.uid')
            ORDER BY cast(json_value(`content`, '$.resourceversion') as bigint) DESC,cast(json_value(`content`, '$.__time_milli__') as bigint) DESC
        ) as rowNum
    FROM `abm-exporter-paimon`.`abm_exporter_db`.`ods_realtime_exporter_lakehouse`
) WHERE rowNum = 1;

CREATE TABLE IF NOT EXISTS `abm-exporter-paimon`.`abm_exporter_db`.`ads_realtime_exporter_lakehouse`(
    kind          varchar,
    cluster      varchar NOT NULL,
    workload_num bigint,
    PRIMARY KEY (kind, cluster) NOT ENFORCED
) with (
    'merge-engine' = 'aggregation',
    'changelog-producer' = 'lookup',
    'fields.workload_num.aggregate-function' = 'sum'
);

insert into `abm-exporter-paimon`.`abm_exporter_db`.`ads_realtime_exporter_lakehouse`
select
    kind,
    cluster,
    case
        when type = 'ADDED' then 1
        when type = 'DELETED' then -1
        ELSE  0
    end as workload_num
from `abm-exporter-paimon`.`abm_exporter_db`.`dwd_realtime_exporter_lakehouse`;

use bigdata_sre;
SET odps.table.scan-options.odps.external.sub.disable.hyper=true;
SET odps.sql.common.table.planner.ext.hive.bridge = true;
set odps.sql.submit.ddltask.via.common.table=true;
SET odps.sql.hive.compatible = false;
set odps.compiler.verify=true;
set odps.isolation.session.enable=true;

-- select * from ods_realtime_exporter_lakehouse limit 100;
-- select * from dwd_realtime_exporter_lakehouse where kind='Node' limit 100;
select * from ads_realtime_exporter_lakehouse order by workload_num desc limit 100;