ApsaraDB for SelectDB:Runtime filter

HashJoinNode
     |                    |
     | 1,000,000          | 200
     |                    |
OlapScanNode         OlapScanNode
     ^                    ^
     | 1,000,000          | 200
    T1 (fact)            T2 (dimension)

HashJoinNode
     |                    |
     | 6,000              | 200
     |                    |
OlapScanNode         OlapScanNode
     ^                    ^
     | 1,000,000          | 200
    T1 (fact)            T2 (dimension)

HashJoinNode
     |                    |
     | 6,000              | 200
     |                    |
OlapScanNode         OlapScanNode
     ^                    ^
     | 6,000              | 200
    T1 (fact)            T2 (dimension)

Conceitos principais

Tabela da esquerda — a tabela no lado esquerdo de um join; usada para a operação de probe. A reordenação de join pode ajustar qual tabela fica em cada lado.

Tabela da direita — a tabela no lado direito de um join; usada para construir a tabela hash. A reordenação de join também pode ajustar essa posição.

Fragment — uma unidade de execução de consulta. O nó frontend (FE) divide uma instrução SQL em fragments e os distribui para os nós backend (BE) no cluster distribuído.

Quando os runtime filters são úteis

Os runtime filters são mais eficazes quando:

• A tabela da esquerda é significativamente maior que a tabela da direita. Gerar um filtro tem custo de memória e computação; isso só compensa em grande escala.

• O resultado do join é muito menor que a tabela da esquerda — o que significa que o filtro consegue descartar a maioria das linhas da tabela da esquerda.

Se a tabela da direita for grande ou se o resultado do join tiver tamanho próximo ao da tabela da esquerda, os runtime filters podem adicionar sobrecarga sem trazer benefícios.

Tipos de filtro

O ApsaraDB for SelectDB oferece suporte a cinco tipos de runtime filter:

Parâmetros

runtime_filter_mode

Controla o escopo de propagação do filtro entre os fragments de execução da consulta.

• **LOCAL** — o produtor do filtro (HashJoinNode) e o consumidor (OlapScanNode) devem estar no mesmo fragment. Adequado para broadcast joins. Menor sobrecarga.

• **GLOBAL** — os filtros são mesclados e transferidos pela rede entre as fronteiras dos fragments. Necessário para shuffle joins em que o produtor e o consumidor estão em fragments diferentes.

O modo GLOBAL cobre todos os cenários atendidos pelo LOCAL, além dos shuffle joins. Mude para LOCAL se a sobrecarga de mesclar e transmitir filtros superar a economia na varredura para um shuffle join específico. Defina como OFF para desativar completamente os runtime filters.

Para obter detalhes sobre o design técnico da mesclagem de filtros entre fragments, consulte ISSUE 6116.

runtime_filter_type

Especifique um tipo ou combine vários tipos:

-- By name (comma-separated, quoted)
SET runtime_filter_type = "BLOOM_FILTER,IN,MIN_MAX";

-- Equivalent numeric form (1=IN, 2=BLOOM_FILTER, 4=MIN_MAX)
SET runtime_filter_type = 7;

Comportamento do predicado IN:

• Usa uma estratégia de mesclagem para execução distribuída.

• Quando o IN e outros tipos de filtro são especificados simultaneamente e a tabela da direita permanece dentro de runtime_filter_max_in_num, o sistema descarta os outros filtros. Como o predicado IN é exato, filtros adicionais não trazem benefícios. Essa otimização aplica-se apenas quando o produtor e o consumidor estão no mesmo fragment.

Comportamento do Bloom filter:

• Possui uma taxa de falsos positivos diferente de zero, o que significa que pode deixar passar algumas linhas não correspondentes. Os resultados permanecem corretos; apenas a eficiência da filtragem é afetada.

• Pode ser enviado ao mecanismo de armazenamento, mas apenas para colunas-chave na tabela da esquerda. Sem o pushdown na camada de armazenamento, o desempenho geralmente piora.

• Evite usar bloom filters em colunas com baixa seletividade ou em tabelas da esquerda pequenas.

Comportamento do filtro MinMax:

• Mais eficaz quando os intervalos de valores das tabelas da direita e da esquerda não se sobrepõem (por exemplo, o máximo da tabela da direita está abaixo do mínimo da tabela da esquerda, ou vice-versa).

• Em colunas numéricas (INT, BIGINT, DOUBLE), intervalos sobrepostos reduzem a eficácia a zero.

• Em colunas não numéricas (VARCHAR), normalmente degrada o desempenho.

Comportamento do IN_OR_BLOOM_FILTER:

• Usa o predicado IN quando as linhas da tabela da direita forem < 102.400; muda para bloom filter acima desse limiar.

• Ajuste o limiar com runtime_filter_max_in_num.

Comportamento do filtro Bitmap:

• Aplicável apenas quando a subconsulta IN retorna uma coluna bitmap.

• Requer o mecanismo vetorizado.

runtime_filter_wait_time_ms

O OlapScanNode aguarda até essa duração por cada runtime filter atribuído antes de iniciar uma varredura. Com três filtros, a espera máxima é três vezes esse valor.

Os filtros que chegam dentro da janela de espera são enviados ao mecanismo de armazenamento. Os filtros que chegam após o início da varredura são aplicados como filtros de expressão nos dados já verificados — eficazes, mas menos eficientes que o pushdown na camada de armazenamento.

Orientações de ajuste:

• Clusters ocupados executando joins longos — aumente este valor para que os filtros tenham tempo de ser construídos e transmitidos antes do início da varredura.

• Clusters leves executando muitas consultas pequenas — diminua este valor para evitar adicionar latência desnecessária a consultas que terminam em segundos.

Se o cluster estiver ocupado com muitas consultas intensivas em recursos ou demoradas, prolongue a duração da espera para evitar que consultas complexas percam oportunidades de otimização. Se a carga do cluster for leve e houver muitas consultas pequenas que levam apenas alguns segundos, reduza a duração da espera para evitar um aumento de 1 s de latência por consulta.

runtime_filters_max_num

Limita o número de bloom filters por consulta. Predicados IN e filtros MinMax não são contabilizados.

Quando o limite é excedido, o sistema retém os bloom filters com maior seletividade — aqueles que devem filtrar mais linhas:

Selectivity = HashJoinNode cardinality / HashJoinNode left child cardinality

Ajuste este parâmetro apenas ao otimizar consultas de join lentas entre tabelas grandes.

Parâmetros de tamanho do bloom filter

O FE calcula o comprimento do bloom filter no momento do planejamento da consulta. Todos os bloom filters do HashJoinNode em uma consulta devem ter o mesmo comprimento para permitir a mesclagem.

• Se a cardinalidade da tabela da direita estiver disponível nas estatísticas, o FE estima o tamanho ideal e arredonda para cima até a potência de 2 mais próxima.

• Se a cardinalidade não estiver disponível, o FE usa runtime_bloom_filter_size como padrão.

• runtime_bloom_filter_min_size e runtime_bloom_filter_max_size limitam o tamanho final, independentemente da estimativa.

Bloom filters maiores lidam com colunas de alta cardinalidade com mais precisão, mas consomem mais memória. Se a precisão da filtragem for insuficiente para uma coluna de alta cardinalidade (milhões de valores distintos), aumente runtime_bloom_filter_size e faça benchmarks do resultado.

Ajuste o tamanho do bloom filter por consulta, não globalmente.

CREATE TABLE test (t1 INT) DISTRIBUTED BY HASH (t1) BUCKETS 2;
INSERT INTO test VALUES (1), (2), (3), (4);

CREATE TABLE test2 (t2 INT) DISTRIBUTED BY HASH (t2) BUCKETS 2;
INSERT INTO test2 VALUES (3), (4), (5);

EXPLAIN SELECT t1 FROM test JOIN test2 WHERE test.t1 = test2.t2;

+-------------------------------------------------------------------+
| Explain String                                                    |
+-------------------------------------------------------------------+
| PLAN FRAGMENT 0                                                   |
|  OUTPUT EXPRS:`t1`                                                |
|                                                                   |
|   4:EXCHANGE                                                      |
|                                                                   |
| PLAN FRAGMENT 1                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test`.`t1`  |
|                                                                   |
|   2:HASH JOIN                                                     |
|   |  join op: INNER JOIN (BUCKET_SHUFFLE)                         |
|   |  equal join conjunct: `test`.`t1` = `test2`.`t2`              |
|   |  runtime filters: RF000[in] <- `test2`.`t2`                   |
|   |                                                               |
|   |----3:EXCHANGE                                                 |
|   |                                                               |
|   0:OlapScanNode                                                  |
|      TABLE: test                                                  |
|      runtime filters: RF000[in] -> `test`.`t1`                    |
|                                                                   |
| PLAN FRAGMENT 2                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test2`.`t2` |
|                                                                   |
|   1:OlapScanNode                                                  |
|      TABLE: test2                                                 |
+-------------------------------------------------------------------+

SET enable_profile = true;

RuntimeFilter:in:
    HasPushDownToEngine: true    -- filter reached the storage engine
    AWaitTimeCost: 0ns           -- no wait; filter arrived before scan started
    EffectTimeCost: 2.76ms       -- time spent applying the filter

RowsVectorPredFiltered: 9,320,008   -- rows discarded by the filter
VectorPredEvalTime: 364.39ms        -- time spent on filter evaluation