Hologres:Distribution key

Como funciona

O Hologres usa a fórmula hash(distribution_key) % shard_count = shard_id para atribuir cada registro a um shard. Para chaves compostas, a fórmula passa a ser hash(col1, col2, ...) % shard_count = shard_id. Todos os registros com o mesmo resultado de hash vão para o mesmo shard.

O princípio fundamental é que operações locais são mais rápidas que operações distribuídas. Quando dados relacionados estão no mesmo shard, o mecanismo evita a redistribuição pela rede e executa toda a computação localmente. Isso gera três ganhos de desempenho:

• Computação paralela: Cada shard processa dados independentemente, permitindo que todos trabalhem em paralelo.

• Pruning de shards: Ao filtrar pela chave de distribuição, o Hologres varre apenas os shards relevantes em vez de todos, melhorando diretamente as consultas por segundo (QPS).

• Local Join: Quando duas tabelas no mesmo grupo de tabelas compartilham a mesma coluna de chave de distribuição na condição de JOIN, os registros correspondentes de ambas ficam no mesmo shard. A junção ocorre inteiramente dentro de cada shard, sem necessidade de movimentação de dados.

Definir uma chave de distribuição

Defina a chave de distribuição na instrução CREATE TABLE. Não é possível alterá-la após a criação da tabela sem recriá-la.

Hologres V2.1 e posterior — use a cláusula WITH:

CREATE TABLE <table_name> (...)
WITH (distribution_key = '<column_name>[,<column_name>]');

Todas as versões — use set_table_property:

BEGIN;
CREATE TABLE <table_name> (...);
CALL set_table_property('<table_name>', 'distribution_key', '<column_name>[,<column_name>]');
COMMIT;

A figura a seguir ilustra a distribuição dos dados entre os shards quando se define uma chave de distribuição.

O número de shards depende da quantidade de nós worker. Para mais informações, consulte Termos.

Escolher uma chave de distribuição

Selecione a chave de distribuição seguindo esta ordem de prioridade:

1. Se houver junção entre tabelas: Defina a coluna do JOIN como chave de distribuição em ambas as tabelas. Em junções com múltiplas tabelas, priorize a maior tabela. Ambas devem pertencer ao mesmo grupo de tabelas. Essa configuração habilita o Local Join e elimina a redistribuição de dados.

2. Se uma coluna for usada frequentemente em GROUP BY: Use essa coluna como chave de distribuição. Os dados serão pré-agregados dentro dos shards, permitindo que consultas com GROUP BY evitem a redistribuição entre shards.

3. Prefira uma coluna de alta cardinalidade: A coluna deve ter uma ampla variedade de valores distintos para garantir uma distribuição uniforme dos dados entre os shards. Uma coluna de baixa cardinalidade — como um indicador de status com poucos estados possíveis — concentra os dados em poucos shards, causando desbalanceamento e sobrecarga.

4. Limite o número de colunas na chave de distribuição: Use no máximo duas colunas. Com uma chave composta, consultas que não incluam todas as colunas da chave ainda podem acionar a movimentação de dados. Evite chaves compostas cujos valores combinados sejam idênticos em muitas linhas, pois isso direciona todas essas linhas ao mesmo shard.

5. Restrição de chave primária: Se a tabela tiver uma chave primária (PK), a chave de distribuição deve ser a própria PK ou um subconjunto de suas colunas. Caso nenhuma chave de distribuição seja especificada, a PK será usada por padrão.

Nota: Para verificar se há desbalanceamento de dados em uma tabela, consulte Visualizar relações de desbalanceamento de workers .

Limitações

• Defina a chave de distribuição na criação da tabela. Para alterá-la, recrie a tabela e reimporte os dados.

• Não é possível atualizar valores nas colunas da chave de distribuição. Para modificar um valor, recrie a tabela.

• Colunas dos seguintes tipos de dados não podem ser usadas como chave de distribuição: Float, Double, Numeric, Array, JSON ou outros tipos complexos.

• Se uma tabela não tiver chave primária, a chave de distribuição pode ficar vazia (sem colunas especificadas), o que distribui os dados aleatoriamente entre os shards. A partir do Hologres V1.3.28, uma chave de distribuição vazia não é mais permitida:

  -- This syntax is prohibited from V1.3.28 onward.
  CALL SET_TABLE_PROPERTY('<table_name>', 'distribution_key', '');

• Um valor null em uma coluna de chave de distribuição é tratado como uma string vazia ("") para fins de hash.

Agregação com GROUP BY

Defina a chave de distribuição como a coluna do GROUP BY para pré-agregar os dados dentro de cada shard. Durante a consulta, o mecanismo evita a redistribuição de dados entre os shards.

Hologres V2.1 e posterior:

CREATE TABLE agg_tbl (
    a int NOT NULL,
    b int NOT NULL
)
WITH (
    distribution_key = 'a'
);

-- Aggregate query: groups by the distribution key column
SELECT a, SUM(b) FROM agg_tbl GROUP BY a;

Todas as versões:

BEGIN;
CREATE TABLE agg_tbl (
    a int NOT NULL,
    b int NOT NULL
);
CALL set_table_property('agg_tbl', 'distribution_key', 'a');
COMMIT;

-- Aggregate query: groups by the distribution key column
SELECT a, SUM(b) FROM agg_tbl GROUP BY a;

Execute EXPLAIN <query> para validar a configuração. Se o plano de execução não contiver um operador de redistribution, os dados permanecem nos shards e não há movimentação entre eles.

JOIN entre duas tabelas

BEGIN;
-- tbl1 distributed by column a, tbl2 distributed by column c.
-- Joining on tbl1.a = tbl2.c: matching records are co-located in the same shard.
CREATE TABLE tbl1 (
    a int NOT NULL,
    b text NOT NULL
)
WITH (
    distribution_key = 'a'
);
CREATE TABLE tbl2 (
    c int NOT NULL,
    d text NOT NULL
)
WITH (
    distribution_key = 'c'
);
COMMIT;

SELECT * FROM tbl1 JOIN tbl2 ON tbl1.a = tbl2.c;

BEGIN;
CREATE TABLE tbl1 (
    a int NOT NULL,
    b text NOT NULL
);
CALL set_table_property('tbl1', 'distribution_key', 'a');

CREATE TABLE tbl2 (
    c int NOT NULL,
    d text NOT NULL
);
CALL set_table_property('tbl2', 'distribution_key', 'c');
COMMIT;

SELECT * FROM tbl1 JOIN tbl2 ON tbl1.a = tbl2.c;

BEGIN;
CREATE TABLE tbl1 (
    a int NOT NULL,
    b text NOT NULL
)
WITH (
    distribution_key = 'a'
);
CREATE TABLE tbl2 (
    c int NOT NULL,
    d text NOT NULL
)
WITH (
    distribution_key = 'd'  -- Mismatched: join is on column c, not d
);
COMMIT;

SELECT * FROM tbl1 JOIN tbl2 ON tbl1.a = tbl2.c;

BEGIN;
CREATE TABLE tbl1 (
    a int NOT NULL,
    b text NOT NULL
);
CALL set_table_property('tbl1', 'distribution_key', 'a');

CREATE TABLE tbl2 (
    c int NOT NULL,
    d text NOT NULL
);
CALL set_table_property('tbl2', 'distribution_key', 'd');
COMMIT;

SELECT * FROM tbl1 JOIN tbl2 ON tbl1.a = tbl2.c;

JOIN entre múltiplas tabelas

Junções envolvendo várias tabelas exigem compensações. Siga estes princípios:

• Mesmo campo de JOIN em todas as tabelas: Defina a coluna compartilhada do JOIN como chave de distribuição para cada tabela.

• Campos de JOIN diferentes: Priorize a junção entre as maiores tabelas. Configure as colunas de JOIN das tabelas grandes como suas chaves de distribuição. Tabelas pequenas contêm menos dados, tornando sua redistribuição menos custosa.

BEGIN;
CREATE TABLE join_tbl1 (
    a int NOT NULL,
    b text NOT NULL
)
WITH (
    distribution_key = 'a'
);
CREATE TABLE join_tbl2 (
    a int NOT NULL,
    d text NOT NULL,
    e text NOT NULL
)
WITH (
    distribution_key = 'a'
);
CREATE TABLE join_tbl3 (
    a int NOT NULL,
    e text NOT NULL,
    f text NOT NULL,
    g text NOT NULL
)
WITH (
    distribution_key = 'a'
);
COMMIT;

-- 3-table join query
SELECT * FROM join_tbl1
INNER JOIN join_tbl2 ON join_tbl2.a = join_tbl1.a
INNER JOIN join_tbl3 ON join_tbl2.a = join_tbl3.a;

BEGIN;
CREATE TABLE join_tbl1 (
    a int NOT NULL,
    b text NOT NULL
);
CALL set_table_property('join_tbl1', 'distribution_key', 'a');

CREATE TABLE join_tbl2 (
    a int NOT NULL,
    d text NOT NULL,
    e text NOT NULL
);
CALL set_table_property('join_tbl2', 'distribution_key', 'a');

CREATE TABLE join_tbl3 (
    a int NOT NULL,
    e text NOT NULL,
    f text NOT NULL,
    g text NOT NULL
);
CALL set_table_property('join_tbl3', 'distribution_key', 'a');
COMMIT;

-- 3-table join query
SELECT * FROM join_tbl1
INNER JOIN join_tbl2 ON join_tbl2.a = join_tbl1.a
INNER JOIN join_tbl3 ON join_tbl2.a = join_tbl3.a;

BEGIN;
-- join_tbl_1: 10 million records, distributed by column a (the large-table JOIN key)
CREATE TABLE join_tbl_1 (
    a int NOT NULL,
    b text NOT NULL
)
WITH (
    distribution_key = 'a'
);
-- join_tbl_2: 1 million records, distributed by column a (matches join_tbl_1)
CREATE TABLE join_tbl_2 (
    a int NOT NULL,
    d text NOT NULL,
    e text NOT NULL
)
WITH (
    distribution_key = 'a'
);
-- join_tbl_3: 1 million records, no distribution key set (small table, redistribution cost is low)
CREATE TABLE join_tbl_3 (
    a int NOT NULL,
    e text NOT NULL,
    f text NOT NULL,
    g text NOT NULL
);
COMMIT;

-- join_tbl_1 and join_tbl_2 use Local Join (keys aligned); join_tbl_3 requires redistribution (small table)
SELECT * FROM join_tbl_1
INNER JOIN join_tbl_2 ON join_tbl_2.a = join_tbl_1.a
INNER JOIN join_tbl_3 ON join_tbl_2.d = join_tbl_3.f;

BEGIN;
-- join_tbl_1: 10 million records
CREATE TABLE join_tbl_1 (
    a int NOT NULL,
    b text NOT NULL
);
CALL set_table_property('join_tbl_1', 'distribution_key', 'a');

-- join_tbl_2: 1 million records
CREATE TABLE join_tbl_2 (
    a int NOT NULL,
    d text NOT NULL,
    e text NOT NULL
);
CALL set_table_property('join_tbl_2', 'distribution_key', 'a');

-- join_tbl_3: 1 million records, no distribution key set
CREATE TABLE join_tbl_3 (
    a int NOT NULL,
    e text NOT NULL,
    f text NOT NULL,
    g text NOT NULL
);
COMMIT;

SELECT * FROM join_tbl_1
INNER JOIN join_tbl_2 ON join_tbl_2.a = join_tbl_1.a
INNER JOIN join_tbl_3 ON join_tbl_2.d = join_tbl_3.f;

Sem chave de distribuição

Sem uma chave de distribuição definida (apenas para tabelas sem chave primária), os dados são distribuídos aleatoriamente entre os shards. Registros com os mesmos valores de campo podem acabar em shards diferentes.

BEGIN;
CREATE TABLE tbl (
    a int NOT NULL,
    b text NOT NULL
);
COMMIT;

A figura a seguir ilustra a distribuição aleatória dos dados quando nenhuma chave de distribuição é definida.

Nota: A partir do Hologres V1.3.28, deixar a chave de distribuição vazia não é permitido para tabelas sem chave primária.

Próximos passos

• Guia baseado em cenários para criação e ajuste de tabelas — escolha as propriedades de tabela adequadas com base nos seus padrões de consulta.

• Otimizar o desempenho de consultas — melhores práticas para ajustar tabelas internas do Hologres.

• Referência DDL para tabelas internas:

  • CREATE TABLE

  • CREATE TABLE AS

  • CREATE TABLE LIKE

  • ALTER TABLE

  • DROP TABLE