Skip to content

feat: vincula Table ao flow que a alimenta (Flow Schedule) - #1112

Merged
Winzen merged 2 commits into
mainfrom
feat/table-flow-schedule-link
Oct 7, 2026
Merged

Winzen merged 2 commits into
mainfrom
feat/table-flow-schedule-link

Conversation

@Winzen

@Winzen Winzen commented Oct 7, 2026

Copy link
Copy Markdown
Contributor

Contexto

As tabelas não tinham nenhuma ligação direta com o flow/deployment do Prefect que as alimenta — último item pendente de basedosdados/pipelines#2065.

Registro completo, test plan e os dois cenários reais testados contra produção: #1111.

Arquivos alterados

backend/apps/admin_data_tools/flow_monitoring.py   # _resolve_tables, _dedicated_table_ids
backend/apps/admin_data_tools/admin.py              # coluna "Tables"
backend/apps/api/v1/models.py                       # Table.flow_schedule, has_flow_schedule
backend/apps/api/v1/admin.py                        # campo "Flow" no form de Table
backend/apps/api/v1/migrations/0057_table_flow_schedule.py
backend/apps/admin_data_tools/tests.py

Mudança

Table.flow_schedule (FK pra DisabledFlowSchedule, related_name="tables"), resolvido em sync-deployments:

def _resolve_tables(name: str, tags: list[str], dedicated_table_ids: set) -> list[Table]:
    candidates = _single_table_candidates(name, tags)
    if candidates:
        table = _resolve_single_table(name, tags)
        return [table] if table else []

    dataset_id = name.removesuffix("_flow")
    tables = Table.objects.filter(cloud_tables__gcp_dataset_id=dataset_id).distinct()
    return [t for t in tables if t.id not in dedicated_table_ids]

dedicated_table_ids é calculado uma vez, olhando todos os deployments do sync antes de vincular qualquer um — garante que uma tabela com flow específico nunca seja "roubada" pelo fallback multi-tabela de outro deployment, independente da ordem em que o Prefect devolve a lista:

def _dedicated_table_ids(deployments: list[dict]) -> set:
    ids = set()
    for dep in deployments:
        table = _resolve_single_table(dep["name"], dep.get("tags") or [])
        if table:
            ids.add(table.id)
    return ids

Benefícios

  • Cada tabela agora mostra, direto no admin, qual flow a alimenta (e vice-versa).
  • Table.has_flow_schedule dá uma checagem rápida, sem denormalizar nada.
  • O fallback multi-tabela cobre flows monolíticos reais (br_me_siconfi_flow, 7 tabelas) sem quebrar quando uma tabela do mesmo dataset foge do padrão e vem de um flow dedicado — confirmado simulando esse cenário real contra produção, não só em teste.

Table ganha flow_schedule (FK pra DisabledFlowSchedule, related_name
"tables") — vínculo automático entre uma tabela e o deployment que a
alimenta, resolvido pelo sync-deployments.

Resolução, em ordem:
- Tag "<dataset_id>__<table_id>" do deployment (deploy_tags(),
  pipelines repo) ou o próprio nome, pra flows antigos que já seguem
  essa convenção direto.
- Sem nenhuma dessas: nome menos o sufixo "_flow", tratado como
  dataset_id puro, vinculando todas as CloudTable daquele dataset de
  uma vez — cobre flow monolítico que alimenta várias tabelas num só
  deployment (confirmado contra produção: br_me_siconfi_flow, 7
  tabelas).

Uma tabela com flow dedicado nunca é roubada pelo fallback acima,
mesmo que o flow monolítico seja processado primeiro no mesmo sync —
a lista inteira de deployments é varrida uma vez antes de qualquer
vínculo, pra achar que tabelas já têm dono específico.

Admin: Table mostra o flow vinculado (label "Flow", clicável);
DisabledFlowSchedule mostra as tabelas que alimenta.

Testado de ponta a ponta contra Postgres real (Docker local): 22
testes passando. Backfill rodado contra produção real (leitura no
Prefect, escrita só no banco local) confirmou os dois cenários reais
que motivaram o design: br_me_siconfi_flow (7 tabelas) e uma tabela
com flow dedicado dentro de um dataset multi-tabela.
@github-actions github-actions Bot added the prod Indica que o Pull Request está com a branch de destino (base) apontando para main label Oct 7, 2026
@Winzen Winzen self-assigned this Oct 7, 2026
Mostra o mesmo campo "Tables" já existente na listagem também na tela
de edição individual, com cada tabela linkando pra sua própria página
de admin.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
@Winzen
Winzen merged commit 1b8e9ef into main Oct 7, 2026
5 checks passed
@Winzen
Winzen deleted the feat/table-flow-schedule-link branch October 7, 2026 22:33
github-actions Bot added a commit that referenced this pull request Oct 7, 2026
@Winzen Winzen linked an issue Oct 7, 2026 that may be closed by this pull request
Winzen added a commit that referenced this pull request Oct 7, 2026
A resolução de tabela por deployment (introduzida em #1112) fazia uma
query de CloudTable por deployment. Com centenas de deployments isso
multiplicou os round-trips ao banco o bastante pra estourar o timeout
do gateway nginx em produção (POST /admin-tools/sync-deployments/
retornando 504 antes de terminar).

Troca por um índice único, montado com uma query no início
(_build_cloud_table_index), e troca o loop de save() por tabela por um
update() em lote por deployment.
Winzen added a commit that referenced this pull request Oct 7, 2026
# Contexto

Depois do merge de #1112, `POST /admin-tools/sync-deployments/` passou a estourar o timeout do gateway nginx (504) em produção, confirmado rodando contra prod via o token já usado por `set_deployment_schedule_active`.

Causa: a resolução de tabela por deployment (`_resolve_single_table`) fazia uma query de `CloudTable` por deployment. Com centenas de deployments, isso multiplicou os round-trips ao banco o bastante pra passar do `proxy_read_timeout` do nginx — confirmado com duas tentativas reais, ambas cortadas em ~61s.

# Mudança

`backend/apps/admin_data_tools/flow_monitoring.py`:

- `_build_cloud_table_index()`: carrega **todos** os `CloudTable` numa única query, indexados em dois dicts — `{(gcp_dataset_id, gcp_table_id): table_id}` pro match direto, `{gcp_dataset_id: {table_id, ...}}` pro fallback monolítico.
- `_resolve_single_table_id`/`_dedicated_table_ids`/`_resolve_table_ids`: agora operam sobre esses dicts em memória (ids, não instâncias de `Table`) em vez de consultar o banco a cada chamada.
- `_sync_deployment`: troca o loop `for table in tables: table.save()` por um único `Table.objects.filter(id__in=table_ids).exclude(...).update(flow_schedule_id=record.id)` por deployment.

Resultado: 1 query de `CloudTable` pro sync inteiro (era N, uma por deployment), e 1 `UPDATE` em lote por deployment com tabelas vinculadas (era N `SELECT` + N `UPDATE`, um por tabela).

# Benefícios

- Sync volta a terminar dentro do timeout do gateway em produção.
- Mesmo comportamento/contrato (`results` dict idêntico), sem mudança de schema.

# Test plan

22 testes automatizados (os mesmos de #1112, incluindo os cenários de flow monolítico e exclusão de tabela dedicada) rodando verde contra Postgres real em Docker local, sem alteração nenhuma nos testes — a refatoração é transparente pro comportamento.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

prod Indica que o Pull Request está com a branch de destino (base) apontando para main

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Vincular Table ao flow que a alimenta (Flow Schedule)

1 participant