Pacote recebido no intake não escrito no BigQuery

Description

Texto padrão para issuetype Bug conforme DoR (Definition of Ready)

01 - PESSOA DE CONTATO (Quem pode responder questões sobre o problema): @Breno Zipoli Monteiro Papa
02 - PROBLEMA (Qual é o problema):

Avaliando o caso da issue https://jiraproducao.totvs.com.br/browse/IDEIA-225, notamos que temos quatro versões do mesmo registro nos dados de intake da tenant em datas diferentes. Especialmente no dia 24-04-2024 vemos que foi o último envio de dados para essa chave na tabela. O registro procurado tem a chave abaixo, para referencias futuras:

  cod_estab = "11"
  AND cod_espec_docto ="DP"
  AND cod_ser_docto ="1"
  AND cod_tit_acr ="0227477"
  AND cod_parcela ="03"

O auditId do último registro enviado com essa chave é 68c856ddabad813d . O valor do campo val_sdo_tit_acr nesse registro é 0.

SELECT * except(counterForEntity), TIMESTAMP_MICROS(CAST(counterForEntity AS INT64)) counterForEntity FROM (
SELECT
  records_landing.* EXCEPT (payload),
  JSON_VALUE(json_payload, '$.val_sdo_tit_acr') val_sdo_tit_acr,
  JSON_VALUE(json_payload, '$.cod_estab') cod_estab,
  JSON_VALUE(json_payload, '$.cod_espec_docto') cod_espec_docto,
  JSON_VALUE(json_payload, '$.cod_ser_docto') cod_ser_docto,
  JSON_VALUE(json_payload, '$.cod_tit_acr') cod_tit_acr,
  JSON_VALUE(json_payload, '$.cod_parcela') cod_parcela,
  JSON_VALUE(attributes, '$.counterForEntity') counterForEntity,
FROM
  labs-app-mdm-production.intake.records_landing records_landing,
  UNNEST(JSON_EXTRACT_ARRAY(payload, "$")) json_payload
WHERE
  publish_time >= '2024-01-01'
  AND tenantId = 'd13dc61c425a4e309987a4853b8ee7a9'
  AND stagingType = 'movfin_tit_acr'
)
WHERE 
  cod_estab = "11"
  AND cod_espec_docto ="DP"
  AND cod_ser_docto ="1"
  AND cod_tit_acr ="0227477"
  AND cod_parcela ="03"

ORDER BY publish_time DESC

bquxjob_2479b64_1904b52f1a9.csv

3.1 KiB

A query abaixo confirma que o mesmo passou pela etapa BIGQUERY_CUSTOMER_WRITER.

select
  *
from
  `labs-app-mdm-production.intake.records_steps`
where
  publish_time >= '2024-04-01'
  and auditId in ('68c856ddabad813d')
  and step like '%BIGQUERY_CUSTOMER_WRITER%'
order by
  publish_time

bquxjob_75073d07_1904a5abbbe.csv

787 B

Porém, podemos ver que na tabela do cliente não temos o registro de auditId 68c856ddabad813d , enviado por último.

SELECT
  val_sdo_tit_acr,
  cod_estab,
  cod_espec_docto,
  cod_ser_docto,
  cod_tit_acr,
  cod_parcela, 
  mdmCreated,
  mdmCounterForEntity__DATETIME__,
  _ingestionDatetime,
  mdmAuditId
FROM
  `carol-d13dc61c425a4e309987.d13dc61c425a4e309987a4853b8ee7a9.ingestion_stg_datasul_carol_movfin_tit_acr`
WHERE
  cod_estab = "11"
  AND cod_espec_docto ="DP"
  AND cod_ser_docto ="1"
  AND cod_tit_acr ="0227477"
  AND cod_parcela ="03"

bquxjob_285a7327_1904b58237f.csv

278 B

Foi feito um teste em outra tenant com um snapshot da tabela para validar se a consolidação de dados poderia ter removido esse registro, simulando a ordem de envio e respeitando os payloads enviados contendo essa chave de registro. Nesse teste tivemos o comportamento esperado (o último registro enviado se manteve).

O teste foi feito nessa tabela, https://daen.carol.ai/datavalidationwebinar/carol-ui/connectors/myconnector/movfin_tit_acr?p=1&ps=25&order=ASC&filters=%5B%7B%22consolidation%22:%22n%22%7D,%7B%22cod_estab%22:%5B%22equal%22,%5B%2211%22%5D%5D%7D,%7B%22cod_espec_docto%22:%5B%22equal%22,%5B%22DP%22%5D%5D%7D,%7B%22cod_ser_docto%22:%5B%22equal%22,%5B%221%22%5D%5D%7D,%7B%22cod_tit_acr%22:%5B%22equal%22,%5B%220227477%22%5D%5D%7D,%7B%22cod_parcela%22:%5B%22equal%22,%5B%2203%22%5D%5D%7D%5D , utilizando a query padrão de consolidação que estava em produção na época.

SELECT val_sdo_tit_acr, * FROM (
SELECT * 
FROM
  `carol-234061fafb784fe98c2d.234061fafb784fe98c2d711f45e3ce7f.ingestion_stg_myconnector_movfin_tit_acr`
WHERE

  STRUCT(mdmId,
    mdmCounterForEntity) NOT IN (
  SELECT
    AS STRUCT mdmId,
    IFNULL(mdmCounterForEntity, 0)
  FROM (
    SELECT
      ROW_NUMBER() OVER (PARTITION BY mdmId ORDER BY mdmCounterForEntity DESC) _RANKING_INTERNAL_FIELD_VIEW_CAROL_,
      *
    FROM
      `carol-234061fafb784fe98c2d.234061fafb784fe98c2d711f45e3ce7f.ingestion_stg_myconnector_movfin_tit_acr`)
  WHERE
    _RANKING_INTERNAL_FIELD_VIEW_CAROL_ = 1
    AND (mdmDeleted = FALSE
      OR mdmDeleted IS NULL) )
) WHERE cod_tit_acr LIKE '%0227477%' and cod_parcela = '03'

Veja que a query de consolidação retiraria os 3 registros com valor de val_sdo_tit_acr (presentes na imagem) e deixaria apenas o último enviado, com valor zerado.

Esses testes levam a crer que o último registro escrito nessa tabela foi o de mdmAuditId = d54674e3bbacaa8b e não o de 68c856ddabad813d , efetivamente o último. Indicando que tivemos problemas na escrita dos dados.

Fazendo uma query para validar se nessa tabela temos mais registros com o mesmo problema, chegamos no total de 186 registros, todos dentro desse mesmo auditId (68c856ddabad813d), onde o payload tinha 786 registros. Foi avaliado o payload e o registro com a chave mencionada é único dentro dele, ou seja, não foi afetado por consolidações dentro do intake (ou não deveria ter sido).

payload.json

3.5 MiB

Iremos recuperar os dados desses 186 registros que não foram ingeridos pela plataforma utilizando o intake, porém é necessário que plataforma verifique se algo aconteceu para corrigir o bug.