fix(world_wb_mides): repair the models, split them per state, and stop table-approve materialising subdirectories

.github/workflows/scripts/prefect_run_dbt.py

@@ -85,10 +85,21 @@ def get_datasets_tables_from_modified_files(
     # generic test or macro, not a table. Treating it as a phantom model would
     # abort prod materialization before any real table builds, so restrict to
     # models.
+    #
+    # A published table is exactly `models/<dataset_id>/<model>.sql`. Anything
+    # deeper is not one: `dataset_id` is read off the parent directory, so
+    # `models/<dataset_id>/<subdir>/<model>.sql` would yield `<subdir>` as the
+    # dataset and launch a materialization flow for a dataset that does not
+    # exist. Subdirectories hold intermediate models (e.g. the ephemeral
+    # per-state models under world_wb_mides/states/, which compile into their
+    # parent and have no table of their own) and archived SQL kept for
+    # reference (br_inpe_queimadas/code/). Skip both.
     sql_files: list[Path] = [
         file
         for file in modified_files
-        if file.suffix == ".sql" and file.parts and file.parts[0] == "models"
+        if file.suffix == ".sql"
+        and len(file.parts) == 3
+        and file.parts[0] == "models"
     ]
 
     datasets_tables: list[tuple[str, str, bool, bool]] = [

dbt_project.yml

@@ -789,6 +789,12 @@ models:
     world_wb_mides:
       +materialized: table
       +schema: world_wb_mides
+      # One model per state feeds each published table. Ephemeral keeps them as
+      # inlined CTEs, so the compiled SQL and the published tables are unchanged
+      # by the split. Switch to `table` (plus a non-published +schema) when the
+      # per-state pipelines need to rebuild one state without touching the rest.
+      states:
+        +materialized: ephemeral
     world_wil_wid:
       +materialized: table
       +schema: world_wil_wid

models/world_wb_mides/states/world_wb_mides__empenho_ce.sql

@@ -0,0 +1,133 @@
+-- Ceará (CE) contribution to world_wb_mides.empenho.
+-- Split out of the monolithic empenho model. The SQL is unchanged: the CTE
+-- bodies are byte-identical and the union order is preserved, so this model
+-- emits exactly the rows and column positions it did before the split.
+-- Column names are positional -- the parent model applies the canonical list.
+-- Materialisation is set in dbt_project.yml (models/world_wb_mides/states).
+with
+    empenhado_ce as (
+        select
+            (safe_cast(extract(year from date(data_emissao_empenho)) as int64)) as ano,
+            (safe_cast(extract(month from date(data_emissao_empenho)) as int64)) as mes,
+            safe_cast(
+                extract(date from timestamp(data_emissao_empenho)) as date
+            ) as data,
+            'CE' as sigla_uf,
+            safe_cast(geoibgeid as string) as id_municipio,
+            safe_cast(codigo_orgao as string) as orgao,
+            safe_cast(trim(codigo_unidade) as string) as id_unidade_gestora,
+            safe_cast(null as string) as id_licitacao_bd,
+            safe_cast(numero_licitacao as string) as id_licitacao,
+            case
+                when tipo_processo_licitatorio = 'N'
+                then '98'
+                when tipo_processo_licitatorio = 'R'
+                then '2'
+                when tipo_processo_licitatorio = 'D'
+                then '8'
+                when tipo_processo_licitatorio = 'I'
+                then '10'
+                when tipo_processo_licitatorio = 'R'
+                then '29'
+            end as modalidade_licitacao,
+            safe_cast(
+                concat(
+                    numero_empenho,
+                    ' ',
+                    trim(codigo_orgao),
+                    ' ',
+                    trim(codigo_unidade),
+                    ' ',
+                    geoibgeid,
+                    ' ',
+                    (substring(data_emissao_empenho, 6, 2)),
+                    ' ',
+                    (substring(data_emissao_empenho, 3, 2))
+                ) as string
+            ) as id_empenho_bd,
+            safe_cast(null as string) as id_empenho,
+            safe_cast(numero_empenho as string) as numero,
+            safe_cast(lower(descricao_empenho) as string) as descricao,
+            safe_cast(modalidade_empenho as string) as modalidade,
+            safe_cast(safe_cast(codigo_funcao as int64) as string) as funcao,
+            safe_cast(safe_cast(codigo_subfuncao as int64) as string) as subfuncao,
+            safe_cast(safe_cast(codigo_programa as int64) as string) as programa,
+            safe_cast(safe_cast(codigo_projeto_atividade as int64) as string) as acao,
+            safe_cast(
+                safe_cast(codigo_elemento_despesa as int64) as string
+            ) as modalidade_despesa,
+            round(safe_cast(valor_empenhado as float64), 2) as valor_inicial,
+        from {{ set_datalake_project("world_wb_mides_staging.raw_empenho_ce") }} e
+
+    ),
+    anulacao_ce as (
+        select
+            safe_cast(
+                concat(
+                    numero_empenho,
+                    ' ',
+                    trim(codigo_orgao),
+                    ' ',
+                    trim(codigo_unidade),
+                    ' ',
+                    geoibgeid,
+                    ' ',
+                    (substring(data_emissao_empenho, 6, 2)),
+                    ' ',
+                    (substring(data_emissao_empenho, 3, 2))
+                ) as string
+            ) as id_empenho_bd,
+            round(sum(safe_cast(valor_anulacao as float64)), 2) as valor_anulacao
+        from {{ set_datalake_project("world_wb_mides_staging.raw_anulacao_ce") }}
+        group by 1
+
+    ),
+    frequencia_ce as (
+        select id_empenho_bd, count(id_empenho_bd) as frequencia_id
+        from empenhado_ce
+        group by 1
+        order by 2 desc
+
+    ),
+    empenho_ce as (
+        select
+            e.ano,
+            e.mes,
+            e.data,
+            e.sigla_uf,
+            e.id_municipio,
+            e.orgao,
+            e.id_unidade_gestora,
+            e.id_licitacao_bd,
+            e.id_licitacao,
+            e.modalidade_licitacao,
+            (
+                case
+                    when frequencia_id > 1
+                    then (safe_cast(null as string))
+                    else e.id_empenho_bd
+                end
+            ) as id_empenho_bd,
+            e.id_empenho,
+            e.numero,
+            e.descricao,
+            e.modalidade,
+            e.funcao,
+            e.subfuncao,
+            e.programa,
+            e.acao,
+            e.modalidade_despesa,
+            round(e.valor_inicial, 2),
+            round(safe_cast(0 as float64), 2) as valor_reforco,
+            round(a.valor_anulacao, 2),
+            round(safe_cast(0 as float64), 2) as valor_ajuste,
+            round(
+                ifnull(e.valor_inicial, 0) - ifnull(a.valor_anulacao, 0), 2
+            ) as valor_final
+        from empenhado_ce e
+        left join frequencia_ce f on e.id_empenho_bd = f.id_empenho_bd
+        full outer join anulacao_ce a on a.id_empenho_bd = e.id_empenho_bd
+
+    )
+select *
+from empenho_ce

models/world_wb_mides/states/world_wb_mides__empenho_df.sql

@@ -0,0 +1,104 @@
+-- Distrito Federal (DF) contribution to world_wb_mides.empenho.
+-- Split out of the monolithic empenho model. The SQL is unchanged: the CTE
+-- bodies are byte-identical and the union order is preserved, so this model
+-- emits exactly the rows and column positions it did before the split.
+-- Column names are positional -- the parent model applies the canonical list.
+-- Materialisation is set in dbt_project.yml (models/world_wb_mides/states).
+with
+    empenho_df as (
+        select
+            (safe_cast(exercicio as int64)) as ano,
+            (safe_cast(extract(month from date(lancamento)) as int64)) as mes,
+            safe_cast(lancamento as date) as data,
+            'DF' as sigla_uf,
+            '5300108' as id_municipio,
+            safe_cast(codigo_ug as string) as orgao,
+            safe_cast(codigo_gestao as string) as id_unidade_gestora,
+            safe_cast(null as string) as id_licitacao_bd,
+            safe_cast(null as string) as id_licitacao,
+            case
+                when codigo_licitacao = '1'
+                then '11'
+                when codigo_licitacao = '2'
+                then '1'
+                when codigo_licitacao = '3'
+                then '2'
+                when codigo_licitacao = '4'
+                then '3'
+                when codigo_licitacao = '5'
+                then '8'
+                when codigo_licitacao = '6'
+                then '10'
+                when codigo_licitacao = '7'
+                then '99'
+                when codigo_licitacao = '8'
+                then '32'
+                when codigo_licitacao = '9'
+                then '4'
+                when codigo_licitacao = '10'
+                then '32'
+                when codigo_licitacao = '11'
+                then '31'
+                when codigo_licitacao = '12'
+                then ''
+                when codigo_licitacao = '13'
+                then '5'
+                when codigo_licitacao = '14'
+                then '6'
+                when codigo_licitacao = '15'
+                then '5'
+                when codigo_licitacao = '16'
+                then '5'
+                when codigo_licitacao = '17'
+                then '6'
+                when codigo_licitacao = '18'
+                then '3'
+                when codigo_licitacao = '19'
+                then '32'
+                when codigo_licitacao = '20'
+                then '31'
+                when codigo_licitacao = '21'
+                then '31'
+                when codigo_licitacao = '22'
+                then '32'
+                when codigo_licitacao = '23'
+                then '12'
+                when codigo_licitacao = '25'
+                then '98'
+                when codigo_licitacao = 'INEXIGÍVEL'
+                then '10'
+            end as modalidade_licitacao,
+            safe_cast(
+                concat(
+                    right(nota_empenho, length(nota_empenho) - 6),
+                    ' ',
+                    codigo_ug,
+                    ' ',
+                    codigo_gestao,
+                    ' ',
+                    '5300108',
+                    ' ',
+                    (right(exercicio, 2))
+                ) as string
+            ) as id_empenho_bd,
+            safe_cast(null as string) as id_empenho,
+            safe_cast(nota_empenho as string) as numero,
+            safe_cast(descricao as string) as descricao,
+            safe_cast(left(modalidade_empenho, 1) as string) as modalidade,
+            safe_cast(cast(codigo_funcao as int64) as string) as funcao,
+            safe_cast(codigo_subfuncao as string) as subfuncao,
+            safe_cast(codigo_programa as string) as programa,
+            safe_cast(codigo_acao as string) as acao,
+            safe_cast(codigo_natureza as string) as elemento_despesa,
+            round(safe_cast(0 as float64), 2) as valor_inicial,
+            round(safe_cast(0 as float64), 2) as valor_reforco,
+            round(safe_cast(0 as float64), 2) as valor_anulacao,
+            round(safe_cast(0 as float64), 2) as valor_ajuste,
+            round(
+                safe_cast(replace (valor_final, ',', '.') as float64), 2
+            ) as valor_final
+        from {{ set_datalake_project("world_wb_mides_staging.raw_empenho_df") }}
+
+    )
+select *
+from empenho_df

models/world_wb_mides/states/world_wb_mides__empenho_mg.sql

@@ -0,0 +1,95 @@
+-- Minas Gerais (MG) contribution to world_wb_mides.empenho.
+-- Split out of the monolithic empenho model. The SQL is unchanged: the CTE
+-- bodies are byte-identical and the union order is preserved, so this model
+-- emits exactly the rows and column positions it did before the split.
+-- Column names are positional -- the parent model applies the canonical list.
+-- Materialisation is set in dbt_project.yml (models/world_wb_mides/states).
+with
+    empenhado_mg as (
+        select
+            safe_cast(ano as int64) as ano,
+            safe_cast(mes as int64) as mes,
+            safe_cast(data as date) as data,
+            'MG' as sigla_uf,
+            safe_cast(id_municipio as string) as id_municipio,
+            safe_cast(trim(orgao) as string) as orgao,
+            safe_cast(id_unidade_gestora as string) as id_unidade_gestora,
+            safe_cast(null as string) as id_licitacao_bd,
+            safe_cast(id_licitacao as string) as id_licitacao,
+            safe_cast(null as string) as modalidade_licitacao,
+            safe_cast(
+                concat(
+                    id_empenho, ' ', orgao, ' ', id_municipio, ' ', (right(ano, 2))
+                ) as string
+            ) as id_empenho_bd,
+            safe_cast(id_empenho as string) as id_empenho,
+            safe_cast(numero_empenho as string) as numero,
+            safe_cast(lower(descricao) as string) as descricao,
+            safe_cast(substring(dsc_modalidade, 5, 1) as string) as modalidade,
+            safe_cast(cast(left(dsc_funcao, 2) as int64) as string) as funcao,
+            safe_cast(cast(left(dsc_subfuncao, 3) as int64) as string) as subfuncao,
+            safe_cast(cast(left(dsc_programa, 4) as int64) as string) as programa,
+            safe_cast(cast(left(dsc_acao, 4) as int64) as string) as acao,
+            safe_cast(
+                replace(left(elemento_despesa, 12), '.', '') as string
+            ) as elemento_despesa,
+            round(safe_cast(valor_empenho_original as float64), 2) as valor_inicial,
+            round(
+                safe_cast(ifnull(safe_cast(valor_reforco as float64), 0) as float64), 2
+            ) as valor_reforco,
+            round(
+                safe_cast(ifnull(safe_cast(valor_anulacao as float64), 0) as float64), 2
+            ) as valor_anulacao,
+            round(safe_cast(0 as float64), 2) as valor_ajuste,
+            round(
+                safe_cast(valor_empenho_original as float64)
+                + safe_cast(ifnull(safe_cast(valor_reforco as float64), 0) as float64)
+                - safe_cast(ifnull(safe_cast(valor_anulacao as float64), 0) as float64),
+                2
+            ) as valor_final
+        from {{ set_datalake_project("world_wb_mides_staging.raw_empenho_mg") }}
+
+    ),
+    dlic as (
+        select
+            id_empenho_bd,
+            case when (count(distinct id_licitacao)) > 1 then 1 else 0 end as dlic
+        from empenhado_mg
+        group by 1
+
+    ),
+    empenho_mg as (
+        select distinct
+            e.ano,
+            e.mes,
+            e.data,
+            e.sigla_uf,
+            e.id_municipio,
+            e.orgao,
+            e.id_unidade_gestora,
+            e.id_licitacao_bd,
+            case
+                when dlic = 1 then (safe_cast(null as string)) else e.id_licitacao
+            end as id_licitacao,
+            e.modalidade_licitacao,
+            e.id_empenho_bd,
+            e.id_empenho,
+            e.numero,
+            e.descricao,
+            e.modalidade,
+            e.funcao,
+            e.subfuncao,
+            e.programa,
+            e.acao,
+            e.elemento_despesa,
+            e.valor_inicial,
+            e.valor_reforco,
+            e.valor_anulacao,
+            e.valor_ajuste,
+            e.valor_final
+        from empenhado_mg e
+        left join dlic l on l.id_empenho_bd = e.id_empenho_bd
+
+    )
+select *
+from empenho_mg