Quando o SQL Basta para Streaming: O Que a Manutenção Incremental de Visões Garante — e O Que Ela Recusa

A resposta padrão para "manter este agregado atualizado sobre um stream" costumava ser um job de stream escrito à mão: definir o estado, escolher os serializadores, conectar os timers, operar o cluster. Bancos de dados de streaming comprimem esse trabalho em uma única instrução. Você escreve CREATE MATERIALIZED VIEW, o engine compila o SQL em um dataflow, e cada evento que chega atualiza o resultado incrementalmente. A promessa de 2026 — RisingWave e Materialize de um lado, o ecossistema Flink SQL convergindo do outro — é que, para read models expressáveis como SQL, o programa DataStream agora é opcional.

Eu queria testar essa promessa em vez de repeti-la. Então construí uma visão no RisingWave, me inscrevi no seu change stream a partir do Node, e então fui procurar os dois pontos em que a promessa se dobra: o estado que a visão acumula silenciosamente e as queries que se recusam a ser incrementais. Estas anotações são o resultado. O engine é o RisingWave porque tem licença Apache e roda localmente em um único container, mas o modelo de custo vem da teoria (DBSP) e se aplica a qualquer engine dessa família.

O que um CREATE MATERIALIZED VIEW de streaming realmente promete

Comece pelo que isso não é. No Postgres, uma visão materializada é um resultado de query em cache. Ela fica desatualizada no momento em que uma tabela base muda, e REFRESH MATERIALIZED VIEW recomputa tudo do zero. A visão é um snapshot que você precisa lembrar de refazer.

Um banco de dados de streaming inverte isso. A definição da visão se torna um grafo de dataflow permanente — sources no topo, operadores com estado no meio, o armazenamento da visão na base. Cada evento flui pelo grafo e toca apenas as linhas que afeta. A visão nunca é atualizada (refreshed) porque nunca está desatualizada por mais do que um intervalo limitado.

A parte interessante é o contrato de consistência, e aqui os detalhes importam mais do que o marketing. O RisingWave injeta barriers em cada source stream — por padrão a cada 1 segundo (barrier_interval_ms = 1000, com um checkpoint a cada barrier). Todos os operadores aplicam as mudanças entre duas barriers atomicamente, ao estilo Chandy-Lamport. Duas consequências derivam desse design, e verifiquei ambas na documentação de arquitetura:

Primeiro, uma leitura nunca observa um evento aplicado pela metade. Se a inserção de um pedido incrementa uma soma por cliente e uma contagem por região na mesma visão, nenhuma query vê a soma atualizada mas a contagem não.

Segundo, visões fazem commit na mesma epoch. Duas visões materializadas derivadas da mesma tabela avançam juntas, então uma query que faz join entre elas não consegue pegar uma visão à frente da outra.

Essa segunda propriedade soa abstrata até você ver o que a ausência dela causa. Em um experimento de 2021, Jamie Brandon executou uma carga de trabalho de razão bancária — 10 milhões de transferências entre 10 contas, de modo que a soma de todos os saldos deve ser sempre zero — através de vários sistemas de streaming e observou as saídas intermediárias. Sistemas que emitem refinamentos por evento sem sincronizar seus streams internos produziram o que ele chamou de "saídas que são completamente impossíveis para qualquer conjunto de entradas". Em uma execução com a Flink Table API, o total emitiu cerca de 38 milhões de updates, e 13.325 deles eram o valor correto: zero. Um quarto de todas as saídas era um único número errado e arbitrário. A implementação em differential-dataflow emitiu exatamente uma saída: zero, uma vez, no primeiro timestamp. Essa propriedade — toda saída é uma resposta correta para algum prefixo consistente da entrada — é chamada de consistência interna, e é toda a razão pela qual engines alinhados por barriers quantizam sua saída da forma como fazem. (O experimento é anterior a várias melhorias do Flink, mas a taxonomia dos modos de falha permanece inalterada; o Materialize, da linhagem do differential-dataflow, vende a mesma propriedade hoje sob seu modo strict-serializable.)

O preço está escrito no mesmo mecanismo: a atualidade (freshness) é quantizada ao intervalo do barrier. Você lê um estado consistente com até cerca de um segundo de idade, não um fio ao vivo por evento. Para um caminho de leitura, considero esse o padrão correto. Um dashboard com um segundo de atraso está ok. Um dashboard que brevemente inventa dinheiro não está.

O modelo de custo: linear, bilinear, holístico

"Incremental" carrega muito peso na expressão manutenção incremental de visões, e o paper do DBSP (VLDB 2023) é o relato mais claro que encontrei sobre o que isso custa. O DBSP prova que qualquer query construída a partir de operadores relacionais pode ser reescrita mecanicamente em um programa sobre deltas. A reescrita sempre existe. O que varia — enormemente — é o estado que cada operador precisa manter para aplicar o próximo delta.

Três camadas cobrem o SQL que eu de fato escrevo:

• Operadores lineares — WHERE, projeções, UNION ALL, transformações por linha. Um delta na entrada produz um delta na saída. Nenhum estado. Estes são gratuitos, no sentido estrito: custo proporcional à mudança, independente do histórico.
• Agregações decomponíveis — SUM, COUNT, AVG via sum/count. Um acumulador por grupo. Um update toca uma linha de estado. MIN e MAX ficam num alçapão dessa camada: em um stream somente de inserções, um valor por grupo basta, mas no momento em que deletes podem chegar, remover o mínimo atual exige conhecer o segundo colocado — então o operador precisa reter todos os valores do grupo.
• Operadores bilineares — joins. A forma incremental de A JOIN B é ΔA ⋈ B + A ⋈ ΔB + ΔA ⋈ ΔB: para processar uma mudança em qualquer um dos lados, o operador precisa do outro lado acumulado. Ambas as entradas são retidas, indefinidamente, a menos que algo as limite.

A camada de join é onde "é só escrever SQL" silenciosamente se torna um exercício de planejamento de capacidade. Um join de três vias entre streams materializa estado intermediário para cada par; cada linha que chega é ao mesmo tempo uma sondagem (probe) no estado dos outros lados e uma adição permanente ao seu próprio. Nada na superfície do SQL te avisa. A definição da visão que se lê como uma query de relatório é, operacionalmente, uma promessa de lembrar dois streams ilimitados para sempre.

Aqui é onde o leitor deve imaginar um update se propagando pelo grafo — o diagrama abaixo contrasta um delta tocando apenas suas linhas de join correspondentes com o rescan de tabela inteira que um refresh em batch realiza.

​

Duas coisas mantêm a conta pagável na prática. A primeira são limites de tempo: um interval join (ON a.ts BETWEEN b.ts - INTERVAL '5' MINUTE AND b.ts) somado a um watermark permite que o engine descarte linhas que não podem mais dar match, convertendo estado ilimitado em uma janela deslizante dele — o artigo sobre streaming joins do RisingWave é explícito ao dizer que esse é o único formato de join com memória previsível. A segunda é onde o estado vive. O RisingWave mantém o estado dos operadores no Hummock, uma LSM tree apoiada em object storage com um cache LRU em memória e em disco local. Um estado de join grande demais degrada em cache misses e latência do S3 em vez de um OOM kill. Esse é um modo de falha melhor, e também mais sutil: o sintoma de um join ilimitado é um gráfico de latência de barrier que sobe lentamente, não um crash.

Minha regra a partir deste estudo: antes de uma visão entrar no caminho de leitura, conte suas entradas de join e nomeie o limite de cada lado — um intervalo de tempo, uma pequena tabela de dimensão, ou uma decisão explícita de que o estado vale sua taxa de crescimento. Se nenhum limite existe, essa é a conversa de design, não o SQL.

Assinando a partir do Node

Uma visão que você só consegue consultar por polling é meio read model. A parte que eu não esperava que o RisingWave acertasse é a subscription: o engine expõe o change stream de cada visão sobre o protocolo puro do Postgres, então um serviço pode consumir inserts, updates e deletes conforme eles fazem commit — sem sink Kafka, sem Debezium, sem peça móvel extra. A documentação de subscription cobre o SQL; a semântica que importa para um consumidor exigiu alguma escavação:

• Um UPDATE em uma linha da visão chega como dois registros compartilhando um mesmo rw_timestamp: um UpdateDelete carregando a linha antiga, depois um UpdateInsert carregando a nova. A coluna op codifica isso — através de um driver você pode receber o enum numérico do data.proto do engine (Insert=1, Delete=2, UpdateInsert=3, UpdateDelete=4), enquanto o psql renderiza os nomes.
• rw_timestamp (milissegundos Unix) é o token de progresso. Persista-o, e um consumidor reiniciado retoma com SINCE <ts> — sem perda, sem duplicatas — desde que o timestamp ainda esteja dentro da janela de retention da subscription.
• A ordenação é garantida entre timestamps diferentes, e dentro de um mesmo timestamp apenas por chave primária. O intercalamento entre chaves diferentes dentro de um barrier é indefinido.
• Desde a v2.1, um fetch pode bloquear no lado do servidor com um timeout, então o loop do consumidor não precisa de sleep-and-poll.

Um script de setup e um consumidor autossuficiente reproduzem tudo acima. O SQL, via psql -h localhost -p 4566 -d dev -U root:

CREATE TABLE orders (order_id INT PRIMARY KEY, customer_id INT, amount DECIMAL);

CREATE MATERIALIZED VIEW customer_totals AS
SELECT customer_id, SUM(amount) AS total_spend, COUNT(*) AS order_count
FROM orders
GROUP BY customer_id;

CREATE SUBSCRIPTION totals_sub FROM customer_totals WITH (retention = '1D');

CREATE TABLE IF NOT EXISTS sub_progress (
  sub_name VARCHAR PRIMARY KEY,
  progress BIGINT
) ON CONFLICT OVERWRITE;

E o consumidor, em um único arquivo:

// subscribe.ts — consome o changefeed de uma visão materializada do RisingWave, com retomada.
import { Client } from "pg";

const SUB = "totals_sub";
// Enum Op do data.proto do RisingWave; o psql mostra nomes, os drivers os números.
const DELETE_OPS = new Set(["2", "4", "Delete", "UpdateDelete"]);

// Réplica local da visão, indexada pela chave primária da visão (customer_id).
const totals = new Map<number, { spend: string; orders: string }>();

async function main(): Promise<void> {
  const db = new Client({ host: "localhost", port: 4566, user: "root", database: "dev" });
  await db.connect();

  const saved = await db.query(
    "SELECT progress FROM sub_progress WHERE sub_name = $1", [SUB],
  );
  const since: string | undefined = saved.rows[0]?.progress;

  // FULL: primeiro o snapshot atual, depois os deltas. SINCE <ts>: retoma, exactly once.
  await db.query(
    since != null
      ? `DECLARE cur SUBSCRIPTION CURSOR FOR ${SUB} SINCE ${since}`
      : `DECLARE cur SUBSCRIPTION CURSOR FOR ${SUB} FULL`,
  );
  console.log(since != null ? `retomando desde ${since}` : "consumindo snapshot completo");

  let unsaved = 0;
  for (;;) {
    // Bloqueia no servidor até uma mudança chegar ou passarem 5s (RisingWave >= 2.1).
    const res = await db.query("FETCH NEXT FROM cur WITH (timeout = '5s')");
    for (const row of res.rows) {
      const key = Number(row.customer_id);
      if (DELETE_OPS.has(String(row.op))) totals.delete(key);
      else totals.set(key, { spend: row.total_spend, orders: row.order_count });
      console.log(`op=${row.op} customer=${key} ->`, totals.get(key) ?? "removido");

      // Linhas de snapshot não carregam rw_timestamp; só os deltas avançam o progresso.
      if (row.rw_timestamp != null && ++unsaved >= 10) {
        await db.query(
          "INSERT INTO sub_progress (sub_name, progress) VALUES ($1, $2)",
          [SUB, row.rw_timestamp],
        );
        await db.query("FLUSH"); // torna a linha de progresso durável antes de confiar nela
        unsaved = 0;
      }
    }
  }
}

main().catch((err) => { console.error(err); process.exit(1); });

Rode com npm install pg tsx && npx tsx subscribe.ts.

Três linhas merecem um passo a passo. O tratamento de DELETE_OPS funciona porque a metade-delete de um update sempre precede sua metade-insert para a mesma chave, então aplicar os registros na ordem de chegada mantém o map local correto — essa é exatamente a ordenação por chave que a documentação garante, e nada mais forte. O guard rw_timestamp != null importa porque um cursor FULL primeiro reproduz o snapshot existente, e as linhas de snapshot não carregam timestamp; fazer checkpoint do progresso durante a fase de snapshot gravaria lixo. E o FLUSH após o upsert de progresso é o que promove de at-least-once para exactly-once entre reinicializações: retome apenas a partir de um progresso que você tornou durável. Insira alguns pedidos pelo psql, atualize um, mate o consumidor no meio do stream, reinicie-o — os pares chegam, a retomada aterrissa exatamente no próximo evento, e o map local converge para o que SELECT * FROM customer_totals mostra.

As queries que se recusam

O resultado do DBSP diz que toda query relacional tem uma forma incremental — não diz que é barata, e engines traçam a linha de formas diferentes. O que respeito na família IVM é que a linha é traçada no momento do CREATE: o planner ou aceita a query e a mantém, ou a rejeita de imediato. A falha é barulhenta e precoce. A alternativa clássica — um cron job em torno de REFRESH MATERIALIZED VIEW — falha silenciosamente, como um SLO de atualidade que se corrói com o tamanho da tabela.

Onde a linha cai, segundo a documentação e minhas próprias tentativas:

• Agregações holísticas. percentile_cont, medianas, qualquer coisa que precise do multiconjunto ordenado completo — um delta não te diz nada sem o resto dos dados. Estes são os verdadeiros casos inviáveis para manutenção incremental; aproximações (sketches ao estilo t-digest) são a saída padrão.
• Ordenação global. Uma janela OVER com PARTITION BY vazio é rejeitada no RisingWave — um ranking global é uma única partição serial, o que estrangularia o dataflow paralelo. De forma relacionada, um ORDER BY isolado na definição de uma visão é aceito mas não mantido; ordenar é tarefa de quem lê.
• Rankings propensos a cascata. RANK() dentro de uma partição é mantido incrementalmente, mas uma inserção pode deslocar todas as linhas abaixo dela — o delta é barato de computar e caro de emitir.
• Diferença de conjuntos e subqueries correlacionadas. EXCEPT e afins são mantidos, mas uma única linha de entrada pode virar linhas de saída arbitrárias; o estado e a amplificação de escrita espelham os de um join.
• Não-determinismo. now() e random() não têm significado incremental coerente; engines ou os rejeitam ou os tratam como caso especial (o RisingWave reescreve predicados baseados em now() como filtros temporais que expiram linhas com o tempo).

Essa lista também é um bom indicador de quando o job escrito à mão paga sua complexidade de volta. Se a lógica precisa de efeitos colaterais por evento, timers de regra de negócio, ou pattern matching ao estilo CEP sobre sequências, não é uma visão — é um programa, e a DataStream API (ou um ator, ou um consumidor com estado explícito) é a ferramenta honesta. O resgate do benchmark do Brandon em 2021 é instrutivo aqui: a solução Flink internamente consistente veio de uma ProcessFunction feita à mão, não da camada SQL. As ferramentas de baixo nível sempre foram suficientes; a questão sempre foi apenas quem compila a query — você ou o engine.

O que eu levo disso

• Trate uma visão materializada de streaming como um dataflow para o qual você planeja capacidade, não como uma query que você coloca em cache. O SQL é a parte fácil; o estado do operador é a conta.
• Antes de colocar uma visão em produção, classifique seus operadores: lineares (gratuitos), agregações decomponíveis (estado por grupo — cuidado com MIN/MAX sob deletes), joins (ambos os lados retidos — nomeie o limite), holísticos (recusados).
• Prefira engines que rejeitam o não-mantível no momento do CREATE a schedulers que recomputam silenciosamente às 3 da manhã.
• Consuma visões através do cursor de subscription, persista rw_timestamp apenas após o FLUSH, e nunca durante a fase de snapshot.
• Leia a latência do barrier como seu indicador antecedente: um barrier lag que sobe é a aparência externa de um estado de join ilimitado.

Recorra à manutenção incremental de visões quando o read model for genuinamente relacional — agregações, joins de enriquecimento com lados limitados, projeções filtradas — e o alvo de atualidade for "cerca de um segundo". Evite-a quando a lógica quer timers, efeitos por evento, ou estatísticas holísticas, ou quando ninguém consegue dizer o que limita o estado do join. Essa última frase é toda a troca em uma linha: o SQL basta para streaming exatamente quando o seu estado basta para o seu SQL.

Fontes: Documentação de subscription do RisingWave · Arquitetura do RisingWave · DBSP, VLDB 2023 · Internal consistency in streaming systems — Jamie Brandon · Understanding streaming joins in RisingWave​