Skip to main content

Linguagem Procedural

No uso de pipelines SQL é possível utilizar comandos procedurais do BigQuery, como variáveis, condicionais, tabelas temporárias e funções temporárias, para obter melhor performance e/ou legibilidade da pipeline.

Na prática, uma pipeline deixa de ser apenas uma única consulta SELECT e passa a ser um script SQL composto por várias instruções executadas em sequência. Isso permite preparar valores, separar etapas intermediárias, reaproveitar resultados e controlar o fluxo de execução antes de retornar o resultado final esperado pela pipeline.

Quando utilizar​

A linguagem procedural é útil quando a query começa a concentrar muita lógica em uma única instrução SQL ou quando algum cálculo intermediário precisa ser reaproveitado em mais de um ponto da pipeline.

Alguns cenários comuns:

  • Definir variáveis para datas, limites de processamento, tenant, batch ou outros parâmetros usados em vários filtros.
  • Melhorar filtros de partição ao calcular previamente um valor e utilizá-lo como constante na consulta principal.
  • Materializar uma etapa intermediária em uma tabela temporária para simplificar JOINs, agregações ou regras de negócio.
  • Executar uma lógica apenas quando houver dados a processar.
  • Reutilizar regras em funções temporárias.

Estrutura básica​

Em scripts procedurais, a pipeline pode ser organizada em etapas. Cada comando é executado sequencialmente, na ordem em que aparece no arquivo.

No exemplo abaixo, primeiro são declaradas e definidas as variáveis de data. Depois, a consulta final utiliza esses valores como filtro. Nesses casos, não esqueça de finalizar cada comando com ;.

DECLARE data_inicio DATE;
DECLARE data_fim DATE;

SET data_inicio = DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY);
SET data_fim = CURRENT_DATE();

SELECT
  device_id,
  status,
  mdmStagingCounter,
  _ingestionDatetime
FROM ingestion_stg_totvs_device
WHERE DATE(_ingestionDatetime) >= data_inicio
  AND DATE(_ingestionDatetime) < data_fim;

Neste exemplo, as datas ficam explícitas no início do script. Isso facilita manutenção e evita repetir a mesma regra em vários pontos da query.

Variáveis com DECLARE e SET​

Variáveis ajudam a tornar a pipeline mais legível e podem melhorar performance em consultas sobre tabelas particionadas.

Em processamentos parciais, é comum buscar o maior mdmStagingCounter já processado no Data Model para ler apenas registros novos da staging table. Quando esse cálculo é feito em uma CTE e reutilizado em vários pontos da query, a mesma regra pode ser recalculada mais de uma vez.

Exemplo usando CTE:

WITH max_counter AS (
  SELECT
    SAFE.DATETIME(TIMESTAMP_MICROS(SAFE_CAST(COALESCE(MAX(mdmStagingCounter), 0) AS INT64))) AS value
  FROM ingestion_device
)

SELECT
  stg.device_id,
  stg.status,
  stg.mdmStagingCounter
FROM ingestion_stg_totvs_device stg
WHERE stg._ingestionDatetime > (SELECT value FROM max_counter)
  AND stg.status IS NOT NULL
UNION ALL
SELECT
  stg.device_id,
  stg.status,
  stg.mdmStagingCounter
FROM ingestion_stg_totvs_device stg
WHERE stg._ingestionDatetime > (SELECT value FROM max_counter)
  AND stg.status IS NULL;

Uma alternativa mais eficiente e legível é calcular esse valor uma única vez em uma variável:

DECLARE max_counter DATETIME;

SET max_counter = (
  SELECT
    SAFE.DATETIME(TIMESTAMP_MICROS(SAFE_CAST(COALESCE(MAX(mdmStagingCounter), 0) AS INT64)))
  FROM ingestion_device
);

SELECT
  stg.device_id,
  stg.status,
  stg.mdmStagingCounter
FROM ingestion_stg_totvs_device stg
WHERE stg._ingestionDatetime > max_counter
  AND stg.status IS NOT NULL
UNION ALL
SELECT
  stg.device_id,
  stg.status,
  stg.mdmStagingCounter
FROM ingestion_stg_totvs_device stg
WHERE stg._ingestionDatetime > max_counter
  AND stg.status IS NULL;

Esse padrão deixa claro qual valor será usado como limite de processamento e evita repetir o cálculo do maior counter sempre que o filtro precisar ser aplicado.

Condicionais com IF​

Condicionais permitem executar partes do script apenas quando uma regra for atendida. Isso é útil para evitar processamentos desnecessários ou para tratar cenários sem dados.

Exemplo:

DECLARE has_new_data BOOL;
DECLARE process_enabled BOOL;

SET has_new_data = (
  SELECT COUNT(1) > 0
  FROM ingestion_stg_totvs_device
  WHERE DATE(_ingestionDatetime) = CURRENT_DATE()
);

SET process_enabled = (
  SELECT
    COUNTIF(setting_value = 'true') > 0
  FROM ingestion_stg_pipeline_settings
  WHERE setting_name = 'process_device'
);

IF has_new_data AND process_enabled THEN
  SELECT
    device_id,
    status,
    mdmStagingCounter,
    _ingestionDatetime
  FROM ingestion_stg_totvs_device
  WHERE DATE(_ingestionDatetime) = CURRENT_DATE();
ELSE
  -- Select vazio para manter a estrutura de retorno da pipeline.
  SELECT
    CAST(NULL AS STRING) AS device_id,
    CAST(NULL AS STRING) AS status,
    CAST(NULL AS INT64) AS mdmStagingCounter,
    CAST(NULL AS DATETIME) AS _ingestionDatetime
  LIMIT 0;
END IF;

Neste exemplo, a consulta principal só é executada quando existem dados novos na staging table e quando o setting process_device está habilitado. Caso contrário, a pipeline retorna um SELECT vazio com a mesma estrutura de campos.

Esse padrão é interessante quando a pipeline precisa manter uma saída previsível, mesmo em cenários sem dados.

Tabelas temporárias​

Tabelas temporárias são criadas durante a execução do script e existem apenas enquanto a execução estiver ativa. Elas ajudam a quebrar queries grandes em etapas menores.

Exemplo:

CREATE TEMP TABLE tmp_active_customers AS
SELECT
  customer_id,
  document_number,
  updated_at
FROM ingestion_stg_totvs_customer
WHERE status = 'ACTIVE';

CREATE TEMP TABLE tmp_orders AS
SELECT
  order_id,
  customer_id,
  total_amount,
  order_date
FROM ingestion_stg_totvs_order
WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY);

SELECT
  o.order_id,
  c.document_number,
  o.total_amount,
  o.order_date
FROM tmp_orders o
INNER JOIN tmp_active_customers c
  ON c.customer_id = o.customer_id;

Esse formato facilita a leitura porque separa a regra de clientes ativos da regra de pedidos recentes. Também evita repetir a mesma subquery várias vezes.

Reutilizando resultados calculados​

Um caso comum de otimização é substituir uma CTE usada várias vezes por uma tabela temporária. A CTE ajuda na legibilidade, mas quando ela é referenciada em mais de um ponto da query, o BigQuery pode precisar recalcular a mesma lógica para cada uso. Se essa lógica lê muitos dados, faz agregações ou possui JOINs custosos, isso pode aumentar o consumo de recursos.

No exemplo abaixo, a CTE customer_metrics é utilizada duas vezes: uma para buscar clientes VIP e outra para calcular um resumo por faixa. Como a mesma regra é consumida em mais de um ponto, o cálculo pode ser repetido.

WITH customer_metrics AS (
  SELECT
    customer_id,
    COUNT(order_id) AS total_orders,
    SUM(total_amount) AS total_spent,
    MAX(order_date) AS last_order_date
  FROM ingestion_stg_totvs_order
  WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 12 MONTH)
  GROUP BY customer_id
),
vip_customers AS (
  SELECT
    customer_id,
    total_orders,
    total_spent
  FROM customer_metrics
  WHERE total_spent >= 10000
),
customer_summary AS (
  SELECT
    CASE
      WHEN total_spent >= 10000 THEN 'VIP'
      WHEN total_spent >= 1000 THEN 'RECORRENTE'
      ELSE 'EVENTUAL'
    END AS customer_range,
    COUNT(1) AS customers
  FROM customer_metrics
  GROUP BY customer_range
)

SELECT
  v.customer_id,
  v.total_orders,
  v.total_spent,
  s.customer_range,
  s.customers
FROM vip_customers v
INNER JOIN customer_summary s
  ON s.customer_range = 'VIP';

Uma alternativa é materializar esse resultado uma única vez em uma tabela temporária e reutilizá-lo nas próximas etapas do script:

CREATE TEMP TABLE tmp_customer_metrics AS
SELECT
  customer_id,
  COUNT(order_id) AS total_orders,
  SUM(total_amount) AS total_spent,
  MAX(order_date) AS last_order_date
FROM ingestion_stg_totvs_order
WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 12 MONTH)
GROUP BY customer_id;

CREATE TEMP TABLE tmp_vip_customers AS
SELECT
  customer_id,
  total_orders,
  total_spent
FROM tmp_customer_metrics
WHERE total_spent >= 10000;

CREATE TEMP TABLE tmp_customer_summary AS
SELECT
  CASE
    WHEN total_spent >= 10000 THEN 'VIP'
    WHEN total_spent >= 1000 THEN 'RECORRENTE'
    ELSE 'EVENTUAL'
  END AS customer_range,
  COUNT(1) AS customers
FROM tmp_customer_metrics
GROUP BY customer_range;

SELECT
  v.customer_id,
  v.total_orders,
  v.total_spent,
  s.customer_range,
  s.customers
FROM tmp_vip_customers v
INNER JOIN tmp_customer_summary s
  ON s.customer_range = 'VIP';

Neste formato, a leitura e agregação de ingestion_stg_totvs_order acontece uma vez na criação de tmp_customer_metrics. Depois disso, as próximas etapas reutilizam a tabela temporária, evitando recalcular a mesma base intermediária e tornando o fluxo mais explícito.

Atenção ao uso de tabelas temporárias

Tabelas temporárias melhoram a organização da pipeline, mas não devem ser usadas sem necessidade. Como os comandos são executados sequencialmente, cada etapa precisa fazer sentido no fluxo da pipeline. Antes de criar uma tabela temporária, avalie se ela realmente simplifica a query ou evita repetição de processamento.

Funções temporárias​

Funções temporárias ajudam a reutilizar regras de transformação dentro da própria pipeline.

Exemplo:

CREATE TEMP FUNCTION normalize_document(document STRING) AS (
  REGEXP_REPLACE(document, r'[^0-9]', '')
);

SELECT
  customer_id,
  normalize_document(document_number) AS normalized_document
FROM ingestion_stg_totvs_customer;

Esse recurso é útil para regras pequenas e reutilizáveis, como normalização de documentos, tratamento de strings, classificação simples ou conversões de formato.

Boas práticas​

  • Mantenha a última consulta compatível com o resultado esperado pela pipeline.
  • Evite gerar múltiplos resultados finais no mesmo script. Prefira preparar dados em variáveis ou tabelas temporárias e deixar apenas um SELECT final.
  • Use variáveis para valores reutilizados ou relevantes para filtros de partição.
  • Use tabelas temporárias quando elas reduzirem repetição, melhorarem legibilidade ou isolarem uma etapa importante da regra.
  • Evite criar muitas etapas intermediárias pequenas, pois isso pode aumentar o tempo total de execução.

Exemplo completo​

O exemplo abaixo mostra uma pipeline de processamento parcial usando o mdmStagingCounter. Primeiro, veja uma versão sem linguagem procedural, em que o cálculo do max_counter fica em uma CTE e é consultado mais de uma vez:

WITH max_counter AS (
  SELECT
    SAFE.DATETIME(TIMESTAMP_MICROS(SAFE_CAST(COALESCE(MAX(mdmStagingCounter), 0) AS INT64))) AS value
  FROM ingestion_device
),
active_devices AS (
  SELECT
    stg.device_id,
    stg.status,
    stg.mdmStagingCounter,
    stg._ingestionDatetime
  FROM ingestion_stg_totvs_device stg
  WHERE stg._ingestionDatetime > (SELECT value FROM max_counter)
    AND stg.status = 'ACTIVE'
),
inactive_devices AS (
  SELECT
    stg.device_id,
    stg.status,
    stg.mdmStagingCounter,
    stg._ingestionDatetime
  FROM ingestion_stg_totvs_device stg
  WHERE stg._ingestionDatetime > (SELECT value FROM max_counter)
    AND stg.status = 'INACTIVE'
)

SELECT * FROM active_devices
UNION ALL
SELECT * FROM inactive_devices;

A versão procedural calcula o maior counter uma vez, guarda esse valor em uma variável e reutiliza nas próximas etapas:

DECLARE max_counter DATETIME;

SET max_counter = (
  SELECT
    SAFE.DATETIME(TIMESTAMP_MICROS(SAFE_CAST(COALESCE(MAX(mdmStagingCounter), 0) AS INT64)))
  FROM ingestion_device
);

CREATE TEMP TABLE tmp_new_devices AS
SELECT
  stg.device_id,
  stg.status,
  stg.mdmStagingCounter,
  stg._ingestionDatetime
FROM ingestion_stg_totvs_device stg
WHERE stg._ingestionDatetime > max_counter;

SELECT
  device_id,
  status,
  mdmStagingCounter,
  _ingestionDatetime
FROM tmp_new_devices
WHERE status = 'ACTIVE'
UNION ALL
SELECT
  device_id,
  status,
  mdmStagingCounter,
  _ingestionDatetime
FROM tmp_new_devices
WHERE status = 'INACTIVE';

As duas versões retornam dados equivalentes. A diferença é que a versão procedural deixa explícito que o max_counter é calculado uma única vez e que a tabela temporária tmp_new_devices concentra os registros novos antes da consulta final.