set 'pipeline.name' = 'bandi';

set 'pipeline.max-parallelism' = '10';
set 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
set 'execution.checkpointing.interval' = '1min';
set 'execution.checkpointing.interval-during-backlog' = '60min';
set 'execution.checkpointing.min-pause' = '1min';
set 'execution.checkpointing.timeout' = '60min';
set 'execution.checkpointing.max-concurrent-checkpoints' = '1';
set 'execution.checkpointing.prefer-checkpoint-for-recovery' = 'true';
set 'state.backend.incremental' = 'true';
set 'table.exec.mini-batch.enabled' = 'true';
set 'table.exec.mini-batch.allow-latency' = '5 s';
set 'table.exec.mini-batch.size' = '5000';
set 'table.exec.sink.upsert-materialize' = 'NONE';

create table if not exists `bandi_source` (
    `partition_id` BIGINT METADATA FROM 'partition' VIRTUAL,
    `id_bando` integer,
    `data_mod` string,
    `be` row<
        `id` integer
    >
) with (
    'connector' = 'kafka',
    'scan.startup.mode' = 'group-offsets',
    'properties.auto.offset.reset' = 'earliest',
    'properties.group.id' = 'flink-bandi',
    'properties.bootstrap.servers' = '...',
    'format' = 'json',
    'topic' = 'bandi_source'
);

create table if not exists `tags_bandi`
(
    `id_bando` integer,
    `tags_id` array<integer>
) with (
    'connector' = 'kafka',
    'scan.startup.mode' = 'group-offsets',
    'properties.auto.offset.reset' = 'earliest',
    'properties.group.id' = 'flink-bandi',
    'properties.bootstrap.servers' = '...',
    'format' = 'json',
    'topic' = 'tags_bandi'
);

create table if not exists `tags_be`
(
    `id_be` integer,
    `be_tags_id` array<integer>
) with (
    'connector' = 'kafka',
    'scan.startup.mode' = 'group-offsets',
    'properties.auto.offset.reset' = 'earliest',
    'properties.group.id' = 'flink-sql-client',
    'properties.bootstrap.servers' = '...',
    'properties.compression.type' = 'gzip',
    'format' = 'json',
    'topic' = 'tags_be'
);

create table `bandi_output` (
    `doc_id` string primary key not enforced,
    `id_bando` integer,
    `data_mod` string,
    `be` row<
        `id` integer,
        `tags_id` array<integer>
    >,
    `tags_id` array<integer>
) with (
    'connector' = 'upsert-kafka',
    'properties.bootstrap.servers' = '...',
    'properties.allow.auto.create.topics' = 'false',
    'properties.compression.type' = 'zstd',
    'properties.max.request.size' = '3145764',
    'key.format' = 'raw',
    'value.format' = 'json',
    'value.fields-include' = 'EXCEPT_KEY',
    'topic' = 'bandi_output'
);

insert into `bandi_output`
select
       coalesce(
               json_string('b-' || cast(`bandi_source`.`id_bando` as string)),
               '"barrier-' || cast(`bandi_source`.`partition_id` as string) || '"'
       ) as `doc_id`,
       `bandi_source`.`id_bando`,
       `bandi_source`.`data_mod`,
       row(
           `bandi_source`.`be`.`id`,
           coalesce(`tags_be`.`be_tags_id`, array[-1])
       ),
       coalesce(`tags_bandi`.`tags_id`, array[-1]) as `tags_id`
from `bandi_source`
        left join `tags_bandi` on `bandi_source`.`id_bando` = `tags_bandi`.`id_bando`
        left join `tags_be` on `tags_be`.`id_be` = `bandi_source`.`stazione_appaltante`.`id`;
