Forem: Pedro Parker

Busca fuzzy em 55 milhões de registros: como pg_trgm salvou meu projeto

Pedro Parker — Fri, 17 Apr 2026 15:37:12 +0000

"Buscar por nome" parece simples até você ter 55 milhões de registros e o usuário esperar resposta em menos de 300ms.

No CNPJ Aberto, os usuários buscam empresas por razão social ("Magazine Luiza"), nome fantasia ("Magalu"), CNPJ ("33.000.167/0001-01"), ou até nome de sócio ("Luiza Helena Trajano"). A busca precisa ser fuzzy (tolerar erros de digitação), rápida, e funcionar em três tabelas diferentes simultaneamente.

Neste post, vou mostrar como fizemos isso sem Elasticsearch — usando apenas PostgreSQL com pg_trgm.

Por que não Elasticsearch?

Elasticsearch é a resposta óbvia para full-text search. Mas:

Mais um serviço para manter, monitorar e escalar
Sincronização entre PostgreSQL e ES é complexa (lag, inconsistências)
Custo — ES consome muita RAM (55M docs = 16+ GB de heap)
Complexidade — para o nosso caso de uso, é um canhão para matar formiga

PostgreSQL com pg_trgm resolve o problema com uma fração da complexidade.

O que é pg_trgm?

pg_trgm (trigram) decompõe strings em conjuntos de 3 caracteres consecutivos e calcula similaridade entre eles.

SELECT show_trgm('Magazine');
-- {"  m"," ma","aga","azi","gaz","ine","mag","ne ","zin"}

Quando você busca "Magazin" (sem o 'e'), o PostgreSQL compara os trigramas e encontra alta similaridade com "Magazine". Isso é o que torna a busca tolerante a erros.

CREATE EXTENSION pg_trgm;

A estrutura das tabelas

A busca precisa cruzar três tabelas:

empresas          → razao_social        (55M linhas)
estabelecimentos  → nome_fantasia       (70M linhas)
socios            → nome_socio          (25M linhas)

Detalhe importante: O resultado final que o usuário vê é uma lista de empresas (identificadas por cnpj_basico). Então mesmo que o match venha da tabela de sócios, precisamos retornar a empresa correspondente.

Os indexes GIN

O coração da performance são os indexes GIN com gin_trgm_ops:

CREATE INDEX CONCURRENTLY ix_empresas_razao_trgm 
ON empresas USING gin (razao_social gin_trgm_ops);

CREATE INDEX CONCURRENTLY ix_estab_fantasia_trgm 
ON estabelecimentos USING gin (nome_fantasia gin_trgm_ops);

CREATE INDEX CONCURRENTLY ix_socios_nome_trgm 
ON socios USING gin (nome_socio gin_trgm_ops);

Por que GIN e não GiST? GIN é mais rápido para leitura (que é 99% do nosso caso). GiST é melhor quando há muitas atualizações. Como nossa base atualiza uma vez por mês, GIN é a escolha certa.

Tamanho dos indexes: cada um ocupa ~3-5 GB. Total de ~12 GB só de indexes trigram. Esse é o tradeoff: espaço em disco por velocidade de busca.

A query: UNION ALL com ILIKE

A estratégia é buscar em cada tabela independentemente e depois unir os resultados:

CANDIDATE_LIMIT = 1000

def search_text(query: str, page: int, per_page: int):
    sql = """
        SELECT cnpj_basico FROM empresas 
        WHERE razao_social ILIKE :pattern
        LIMIT :lim

        UNION ALL

        SELECT cnpj_basico FROM estabelecimentos 
        WHERE nome_fantasia ILIKE :pattern
        LIMIT :lim

        UNION ALL

        SELECT cnpj_basico FROM socios 
        WHERE nome_socio ILIKE :pattern
        LIMIT :lim
    """

    candidates = db.execute(sql, {
        "pattern": f"%{query}%", 
        "lim": CANDIDATE_LIMIT
    })

    # Deduplica e pagina
    unique_cnpjs = list(dict.fromkeys(
        row.cnpj_basico for row in candidates
    ))
    page_cnpjs = unique_cnpjs[(page-1)*per_page : page*per_page]

    # Hydrata com dados de exibição
    return hydrate(page_cnpjs)

Por que ILIKE e não o operador `%` (similarity)?

O operador % do pg_trgm (WHERE razao_social % 'Magazine') calcula similaridade e retorna matches acima de um threshold. É ótimo para fuzzy search puro, mas:

ILIKE '%query%' + GIN trigram é surpreendentemente eficiente — o PostgreSQL usa o index GIN para filtrar candidatos pelo trigram e depois aplica o ILIKE como filtro final
ILIKE dá matches exatos de substring, que é o que o usuário geralmente quer
O threshold do % precisa de tuning fino e pode retornar resultados irrelevantes

O LIMIT por branch

Cada branch do UNION tem LIMIT 1000. Isso é crucial:

Sem limit, uma query genérica como "COMERCIO" varreria milhões de linhas
Com limit, mesmo o pior caso retorna em < 500ms
O cap total (MAX_TOTAL_VISIBLE = 1000) garante que nunca paginamos além de resultados relevantes

Buscas numéricas: atalho por CNPJ

Quando a query é numérica, pulamos o trigram e vamos direto:

def search_cnpj(digits: str):
    if len(digits) >= 8:
        # CNPJ basico completo — equality, instantâneo
        return db.query(Empresa).filter(
            Empresa.cnpj_basico == digits[:8]
        ).all()
    else:
        # CNPJ parcial — prefix match
        return db.query(Empresa).filter(
            Empresa.cnpj_basico.like(f"{digits}%")
        ).limit(MAX_TOTAL_VISIBLE).all()

B-tree index no cnpj_basico (PK) = resposta em < 1ms.

Cache com Redis

Buscas textuais são caras (3 scans trigram + dedup + hydrate). Cache é obrigatório:

import json, hashlib

def search_with_cache(query, page, per_page):
    cache_key = f"search:{query}:{page}:{per_page}"

    cached = redis.get(cache_key)
    if cached:
        return json.loads(cached)

    results = search_text(query, page, per_page)

    # Cache por 5 minutos — dados mudam mensalmente, 
    # mas consultas novas precisam aparecer rápido
    redis.setex(cache_key, 300, json.dumps(results))

    return results

Hit rate observado: ~40% em produção. Queries populares ("Petrobras", "Magazine Luiza") são servidas do cache.

Hydratação: evitando N+1

Depois de obter a lista de cnpj_basico da busca, precisamos carregar razão social, CNPJ completo e situação cadastral para exibição. Fazer uma query por resultado seria N+1:

def hydrate(cnpj_basicos: list[str]):
    # Uma única query com IN clause
    empresas = db.query(Empresa).filter(
        Empresa.cnpj_basico.in_(cnpj_basicos)
    ).all()
    empresa_map = {e.cnpj_basico: e for e in empresas}

    # Buscar estabelecimento matriz para cada empresa
    matrizes = db.query(Estabelecimento).filter(
        Estabelecimento.cnpj_basico.in_(cnpj_basicos),
        Estabelecimento.identificador_matriz_filial == "1"
    ).all()
    matriz_map = {m.cnpj_basico: m for m in matrizes}

    results = []
    for cnpj in cnpj_basicos:
        emp = empresa_map.get(cnpj)
        mat = matriz_map.get(cnpj)
        if emp and mat:
            results.append({
                "cnpj": format_cnpj(mat.cnpj_basico, mat.cnpj_ordem, mat.cnpj_dv),
                "razao_social": emp.razao_social,
                "situacao_cadastral": mat.situacao_cadastral,
            })

    return results

2 queries no total, independente do número de resultados.

Resultados de performance

Query	Tempo (sem cache)	Tempo (com cache)
"Petrobras"	~120ms	~2ms
"Comércio varejista"	~350ms	~2ms
"33000167000101"	~5ms	~2ms
"João Silva" (sócio)	~280ms	~2ms

Tudo isso rodando em um PostgreSQL 16 com 8 GB de RAM. Sem Elasticsearch, sem Meilisearch, sem nada além do PostgreSQL.

Conclusão

pg_trgm é subestimado. Para aplicações onde:

O dataset cabe em PostgreSQL (até centenas de milhões de registros)
A busca é por substring/similaridade, não full-text semântico
Você quer manter a stack simples

...ele é a escolha certa. Adicione cache no Redis para queries repetidas e você tem um search engine que aguenta muito tráfego com hardware modesto.

Quer ver funcionando na prática? Teste uma busca no CNPJ Aberto e digite qualquer nome de empresa ou CNPJ e veja o autocomplete em tempo real.

Como importei 55 milhões de empresas para PostgreSQL em menos de 3 horas

Pedro Parker — Fri, 17 Apr 2026 00:29:21 +0000

Quando decidi construir o CNPJ Aberto, que é uma plataforma gratuita de consulta de empresas brasileiras, o primeiro desafio foi óbvio: como colocar 55 milhões de registros dentro do PostgreSQL de forma rápida e repetível (a base atualiza todo mês)?

Os dados vêm dos Dados Abertos da Receita Federal, distribuídos em dezenas de ZIPs com CSVs em latin-1, separados por ;, com campos inconsistentes (muito inconsistentes!!).

A ideia deste post rápido é mostrar as técnicas que transformaram uma importação de 12+ horas em menos de 3 horas.

O cenário

Tabela	Registros aproximados	Peso CSV
`empresas`	~55M	~4 GB
`estabelecimentos`	~70M	~15 GB
`socios`	~25M	~3 GB
`simples`	~35M	~2 GB
Tabelas auxiliares	~15K total	<1 MB

Total: ~25 GB de CSVs descompactados, distribuídos em ~40 arquivos ZIP. Nestes ponto é visível que os maiores problemas seriam empresas e estabelecimentos, não só para querys como para joins, que são muitos para criar um sistema legal e atrativo.

Tentativa 1: INSERT com ORM (12+ horas)

A abordagem ingênua com SQLAlchemy:

for row in csv_reader:
    empresa = Empresa(**parse_row(row))
    session.add(empresa)
    if i % 1000 == 0:
        session.commit()

Resultado: ~1.200 inserts/segundo. Para 55M de registros, isso dá ~12 horas só para a tabela empresas. Inaceitável. Eu acabei ingerindo tudo pois imaginava que > 12 horas de espera seirma melhores que > X horas de implementação e pesquisa para melhoria de queries. Talvez fossem, mas pelo desafio, fui além.

O ORM adiciona overhead em cada objeto: validação de tipo, tracking de estado, construção de SQL dinâmico.

Para melhorias, a IA acaba ajudando um pouco, não muito, ela alucinada bastante na ajuda e as vezes acaba piorando querys e criando index sem sentido, usei o Opus 4.6. Muito útil, mas o double check é necessário.

Tentativa 2: `executemany` com batches (4+ horas)

Removendo o ORM e usando psycopg2 direto:

BATCH_SIZE = 5000
batch = []
for row in csv_reader:
    batch.append(parse_row(row))
    if len(batch) >= BATCH_SIZE:
        cursor.executemany(INSERT_SQL, batch)
        conn.commit()
        batch.clear()

Resultado: ~4.000 inserts/segundo. Melhor, mas ainda 4+ horas.

O problema: executemany ainda gera um INSERT por linha. O PostgreSQL parseia e planeja cada statement individualmente.

Solução final: COPY + Temp Tables (< 3 horas)

O COPY é o mecanismo de bulk load nativo do PostgreSQL. Ele bypassa o parser SQL, o planner e o executor — escrevendo direto no heap da tabela. É 10-50x mais rápido que INSERT.

Passo 1: Otimizar a sessão

cursor.execute("SET synchronous_commit = off")
cursor.execute("SET work_mem = '256MB'")

synchronous_commit = off permite que o PostgreSQL confirme transações sem esperar o flush do WAL para disco. Seguro para data loads (se o servidor crashar, você reimporta). Como nesse caso os dados não são tão importantes, é possível reimportar sem precisar fazer check de nada.

Passo 2: Dropar indexes antes, recriar depois

Indexes tornam cada INSERT mais caro porque o B-tree/GIN precisa ser atualizado. Para bulk load, é mais eficiente dropar tudo, importar, e recriar:

def drop_indexes(cursor):
    cursor.execute("""
        SELECT indexname, tablename FROM pg_indexes 
        WHERE schemaname = 'public' 
        AND indexname NOT LIKE '%_pkey'
    """)
    for idx, table in cursor.fetchall():
        cursor.execute(f"DROP INDEX IF EXISTS {idx}")

def create_indexes(cursor):
    # Recriar com CONCURRENTLY para não bloquear reads
    cursor.execute("""
        CREATE INDEX CONCURRENTLY IF NOT EXISTS 
        ix_empresas_razao_trgm ON empresas 
        USING gin (razao_social gin_trgm_ops)
    """)
    # ... mais indexes

CREATE INDEX CONCURRENTLY é crucial — permite que o site continue respondendo enquanto os indexes são construídos.

Passo 3: COPY via temp table + UPSERT

Para tabelas que precisam de upsert (atualização mensal), usamos temp tables:

BATCH_SIZE = 200_000

def import_batch(cursor, table, columns, rows):
    # 1. Criar temp table com mesma estrutura
    cursor.execute(f"CREATE TEMP TABLE tmp_{table} (LIKE {table} INCLUDING DEFAULTS)")

    # 2. COPY os dados para a temp table
    csv_buffer = io.StringIO()
    writer = csv.writer(csv_buffer)
    for row in rows:
        writer.writerow(row)
    csv_buffer.seek(0)

    cursor.copy_expert(
        f"COPY tmp_{table} ({','.join(columns)}) FROM STDIN WITH (FORMAT csv, NULL '')",
        csv_buffer
    )

    # 3. UPSERT da temp table para a tabela real
    cols = ', '.join(columns)
    update_cols = ', '.join(f"{c} = EXCLUDED.{c}" for c in columns if c != pk)

    cursor.execute(f"""
        INSERT INTO {table} ({cols})
        SELECT {cols} FROM tmp_{table}
        ON CONFLICT ({pk}) DO UPDATE SET {update_cols}
    """)

    # 4. Limpar
    cursor.execute(f"DROP TABLE tmp_{table}")

Por que temp table? Porque o COPY não suporta ON CONFLICT diretamente. A temp table recebe o bulk load ultrarrápido, e depois um único INSERT ... ON CONFLICT faz o merge.

Passo 4: Paralelismo no download e importação

Os ZIPs da Receita Federal são independentes, então podemos baixar e importar em paralelo:

with ThreadPoolExecutor(max_workers=4) as pool:
    futures = []
    for zip_url in zip_urls:
        futures.append(pool.submit(download_and_import, zip_url))

    for future in as_completed(futures):
        future.result()  # propaga exceções

4 workers = 4 ZIPs sendo importados simultaneamente. Com SSDs, o PostgreSQL lida bem com escritas paralelas em tabelas diferentes.

Resultados

Abordagem	Velocidade	Tempo total
ORM (SQLAlchemy)	~1.200/s	12+ horas
`executemany` batches	~4.000/s	4+ horas
`COPY` + temp tables + parallelismo	~80.000/s	< 3 horas

67x mais rápido que a abordagem inicial.

Lições aprendidas

use COPY para bulk load no PostgreSQL. Não existe nada mais rápido sem ir para pg_bulkload ou extensões externas.
Dropar indexes antes de um bulk load e recriar depois é quase sempre mais rápido que manter os indexes durante a carga.
Temp tables são o bridge entre COPY (que não suporta upsert) e a necessidade de ON CONFLICT.
synchronous_commit = off é uma otimização segura para data loads — o pior que acontece é perder dados que você pode reimportar.
Batch size importa: 200K linhas por batch é o sweet spot. Muito menos = overhead de transação. Muito mais = uso excessivo de memória.

O resultado

Esse pipeline roda todo mês no CNPJ Aberto para atualizar a base com os dados mais recentes da Receita Federal. Qualquer pessoa pode consultar gratuitamente dados de qualquer empresa brasileira, razão social, sócios, endereço, CNAE, situação cadastral e muito mais.

Se você trabalha com dados públicos brasileiros, dá uma olhada: cnpjaberto.com.br

Obrigado!