ApsaraDB for SelectDB:Usar Bucket Shuffle Join

Funcionamento

O ApsaraDB for SelectDB oferece suporte a duas estratégias comuns de Join distribuído:

• Broadcast Join: envia a tabela da direita completa para cada HashJoinNode que contém dados da tabela da esquerda. Com três HashJoinNodes, isso representa 3x o volume de dados da tabela da direita, tanto em rede quanto em memória.

• Shuffle Join: aplica hash em ambas as tabelas e redistribui todas as linhas pelos nós do cluster. A sobrecarga de rede equivale ao tamanho combinado das duas tabelas; a sobrecarga de memória corresponde ao tamanho da tabela da direita.

O nó frontend (FE) armazena as informações de distribuição de dados de cada tabela. Quando a condição de igualdade de um Join corresponde à coluna de bucket da tabela da esquerda, o FE utiliza esse mapa de distribuição para enviar os dados da tabela da direita diretamente aos nós de armazenamento da tabela da esquerda, sem necessidade de redistribuir a tabela da esquerda.

Como resultado, a sobrecarga de rede e de memória equivale apenas ao tamanho da tabela da direita. Essa vantagem aumenta quando o FE também consegue aplicar partition pruning ou bucket pruning na tabela da esquerda, o que reduz ainda mais o volume de dados que cada nó precisa processar.

Diferença entre Bucket Shuffle Join e Colocate Join: O Bucket Shuffle Join não exige conhecimento prévio nem alterações na distribuição de dados de nenhuma das tabelas. Além disso, evita data skew, pois não impõe regras de distribuição correspondentes entre as duas tabelas. Ele também oferece mais caminhos para o Join Reorder explorar durante o planejamento da consulta.

Conceitos principais

Ative o Bucket Shuffle Join

Defina a variável de sessão enable_bucket_shuffle_join como true. O FE selecionará automaticamente o Bucket Shuffle Join para qualquer consulta elegível.

set enable_bucket_shuffle_join = true;

Ao planejar uma consulta distribuída, o FE segue esta ordem de prioridade:

Colocate Join > Bucket Shuffle Join > Broadcast Join > Shuffle Join

Para substituir a seleção automática, utilize um hint:

SELECT * FROM test JOIN [shuffle] baseall ON test.k1 = baseall.k1;

Verifique o uso do Bucket Shuffle Join

Execute explain em uma consulta para inspecionar seu plano de execução. Procure por BUCKET_SHUFFLE no campo join op.

Bucket Shuffle Join aplicado:

|   2:HASH JOIN                                                       |
|   |  join op: INNER JOIN (BUCKET_SHUFFLE)                          |
|   |  hash predicates:                                              |
|   |  colocate: false, reason: table not in the same group          |
|   |  equal join conjunct: `test`.`k1` = `baseall`.`k1`            |

Retorno a outro método de Join (Bucket Shuffle Join não foi aplicado):

|   2:HASH JOIN                                                       |
|   |  join op: INNER JOIN (BROADCAST)                               |
|   |  hash predicates:                                              |
|   |  colocate: false, reason: table not in the same group          |
|   |  equal join conjunct: `test`.`k1` = `baseall`.`k1`            |

Caso o plano exiba BROADCAST ou SHUFFLE em vez de BUCKET_SHUFFLE, revise as regras de planejamento abaixo para identificar o motivo pelo qual a otimização não foi aplicada.

Regras de planejamento

O Bucket Shuffle Join roteia dados com base em atribuições de bucket calculadas por hash. Todos os requisitos de planejamento derivam dessa restrição central: o FE deve ser capaz de determinar, no momento do planejamento, exatamente quais nós de armazenamento contêm cada linha da tabela da esquerda.