PolarDB:Use Hash Match to perform IMCI-enabled operations

doBuild

Cada thread de trabalho constrói tabelas hash independentes para cada partição usando os dados da tabela esquerda, produzindo uma matriz worker x partition de mapas hash:

Cada entrada da tabela hash é um valor UInt64 que codifica a posição do chunk correspondente àquela chave. O valor UInt64 divide-se em três campos no nível de bits:

• UInt16: a thread de trabalho proprietária do chunk

• UInt16: o deslocamento do chunk

• UInt32: o índice do array dentro do chunk

As threads de trabalho buscam tuplas da tabela esquerda em paralelo e as inserem na tabela hash designada para seu worker e partição, sem bloqueio. Esse processo continua até que todos os dados da tabela esquerda sejam processados.

doMerge

Após a execução de doBuild, cada partição possui uma tabela hash por thread de trabalho. Como a fase de sondagem requer uma única tabela hash por partição, doMerge consolida todas as tabelas hash por worker de uma partição em uma só:

Uma única thread de trabalho mescla cada partição de forma independente, eliminando a sincronização por bloqueio. Como o número de partições excede o número de threads de trabalho, a carga distribui-se uniformemente entre os workers.

Transbordamento para disco na fase de construção

Cada thread de trabalho recebe uma parcela igual da cota total de memória. Quando a memória disponível não comporta todas as tabelas hash e chunks, o operador descarrega partições inteiras para o disco, começando pela partição de maior número, até que as partições restantes na memória possam ser processadas normalmente. Se nenhuma partição couber na memória, o sistema lança uma exceção OutOfMemoryError.

Durante a execução de doBuild, se uma thread de trabalho não conseguir adicionar um par chave-valor a uma tabela hash ou armazenar um chunk devido à memória insuficiente, ela descarrega os dados de sua partição de maior número para o disco:

1. Os chunks são gravados em um arquivo temporário e sua memória é liberada.

2. A tabela hash dessa partição é excluída.

3. Quando a partição for necessária posteriormente, os chunks serão lidos do arquivo temporário e a tabela hash será reconstruída.

O maior número de partição descarregada é visível para todas as threads de trabalho. Se a partição descarregada mais alta de um worker exceder o máximo global atual, ele atualiza o valor global e libera a memória correspondentemente. Para partições acima do maior número atual em memória, as threads de trabalho ignoram a construção de tabelas hash e, em vez disso, gravam dados diretamente em chunks, descarregando para o disco quando um chunk estiver cheio.

doFetch

As threads de trabalho buscam dados da tabela direita e procuram tuplas correspondentes na tabela hash da partição de cada linha. As tuplas correspondidas são processadas de acordo com o tipo de junção. Esse ciclo se repete até que todas as threads de trabalho consumam a tabela direita.

Transbordamento para disco na fase de sondagem

Quando a fase de construção descarrega algumas partições para o disco, a fase de sondagem trata as partições em memória e em disco separadamente.

Durante a execução de doFetch, após uma thread de trabalho buscar uma tupla da tabela direita:

• Se a partição da tupla estiver na memória, o worker pesquisa diretamente na tabela hash.

• Caso a partição esteja no disco, a tupla é armazenada no chunk dessa partição. Quando o chunk estiver cheio, ele é liberado para o disco e a memória do chunk é liberada.

Depois que todos os workers terminam de ler a tabela direita, as partições em memória são processadas e sua memória é liberada. Em seguida, as partições em disco são processadas uma por vez, cada uma tratada independentemente por uma única thread de trabalho (sem necessidade de bloqueio). Como o número de partições excede o número de threads de trabalho, a carga de trabalho permanece equilibrada.

O processamento de cada partição em disco segue o mesmo padrão de construção e depois sondagem:

1. Construção: As threads de trabalho leem os dados da partição da tabela esquerda dos arquivos temporários, desserializam os chunks e constroem a tabela hash.

2. Sondagem: As threads de trabalho leem os dados da partição da tabela direita dos arquivos temporários, desserializam os chunks e buscam correspondências.

Após o processamento de todas as partições, o operador Hash Match é concluído. Embora o processamento de partições em memória e em disco seja descrito separadamente aqui, ambos os caminhos compartilham uma única implementação de código unificada.

Etapas do processo de sondagem

Todas as operações de sondagem passam por três etapas, cada uma executada pela função probe:

1. ProbeMem: Lê os dados da tabela direita e encaminha cada tupla para processamento em memória (busca na tabela hash) ou armazenamento em disco (gravação em arquivo temporário para processamento posterior no ProbeDisk).

2. ProbeLeft: Trata variantes de LEFT JOIN (LEFT OUTER JOIN, LEFT SEMI JOIN, LEFT ANTI SEMI JOIN). Percorre todos os pares chave-valor na tabela hash e filtra tuplas correspondidas ou não correspondidas.

3. ProbeDisk: Processa partições em disco uma por vez, carregando chunks de arquivos temporários e executando a função probe. Para operações LEFT JOIN, a lógica de JOIN também é aplicada a cada partição de disco.

Inner

Quando uma tupla da tabela direita é não NULL e corresponde à tabela hash construída a partir da tabela esquerda, ambas as tuplas (esquerda e direita) são emitidas.

LeftOuter

Se uma tupla da tabela direita for não NULL e houver correspondência, ambas as tuplas (esquerda e direita) são emitidas. Tuplas da tabela esquerda sem correspondência são emitidas com NULL nas posições da tabela direita. Caso exista um PostFilter, as entradas correspondidas da tabela hash são verificadas pelo PostFilter antes da saída.

RightOuter

Se uma tupla da tabela direita for não NULL e houver correspondência, ambas as tuplas (esquerda e direita) são emitidas. Tuplas da tabela direita sem correspondência são emitidas com NULL nas posições da tabela esquerda. Caso exista um PostFilter, as entradas correspondidas da tabela hash são verificadas pelo PostFilter antes da saída.

LeftSemi

O LEFT SEMI JOIN segue uma estrutura semelhante ao LEFT OUTER JOIN, mas apenas as tuplas da tabela esquerda são emitidas, não as da tabela direita. A saída depende dos resultados de correspondência e se semi_probe_ está definido, conforme esta tabela verdade:

//+------------------------------+--------------+----------------+
//|           matched            | semi_probe_  | ! semi_probe_  |
//+------------------------------+--------------+----------------+
//| normal true                  | (left, TRUE) |   (left, ONLY) |
//+------------------------------+--------------+----------------+
//+------------------------------+--------------+----------------+
//|        ! matched             | semi_probe_  | ! semi_probe_  |
//+------------------------------+--------------+----------------+
//|NULL v.s. (empty set)         |              |                |
//|e.g., NULL IN (empty set)     | (left, FALSE)|  NO_OUTPUT     |
//+------------------------------+--------------+----------------+
//|NULL v.s. (set)               |              |                |
//|e.g., NULL IN (1, 2, 3)       | (left, NULL) |  NO_OUTPUT     |
//+------------------------------+--------------+----------------+
//|left_row v.s. (set with NULL) |              |                |
//|e.g., 10 IN (1, NULL, 3)      | (left, NULL) |  NO_OUTPUT     |
//+------------------------------+--------------+----------------+
//|normal false                  |              |                |
//|e.g., 10 IN (1, 2, 3)         | (left, FALSE)|  NO_OUTPUT     |
//+------------------------------+--------------+----------------+

Caso exista um PostFilter, as entradas correspondidas da tabela hash são verificadas pelo PostFilter.

LeftAntiSemi

O LEFT ANTI SEMI JOIN é semelhante ao LEFT OUTER JOIN. Apenas as tuplas da tabela esquerda são emitidas, com base na seguinte tabela verdade:

//+------------------------------+----------------+
//|        ! matched             | ! semi_probe_  |
//+------------------------------+----------------+
//|NULL v.s. (empty set)         |                |
//|e.g., NULL NOT IN (empty set) |   (left, ONLY) |
//+------------------------------+----------------+
//|NULL v.s. (set)               |                |
//|e.g., NULL NOT IN (1, 2, 3)   |   (left, ONLY) |
//+------------------------------+----------------+
//|left_row v.s. (set with NULL) |                |
//|e.g., 10 NOT IN (1, NULL, 3)  |   (left, ONLY) |
//+------------------------------+----------------+
//|normal false                  |                |
//|e.g., 10 NOT IN (1, 2, 3)     |   (left, ONLY) |
//+------------------------------+----------------+

Caso exista um PostFilter, as entradas correspondidas da tabela hash são verificadas pelo PostFilter.

RightSemi

O RIGHT SEMI JOIN é semelhante ao RIGHT OUTER JOIN. Apenas as tuplas da tabela direita são emitidas, com base nesta tabela verdade:

//+------------------------------+--------------+----------------+
//|           matched            | semi_probe_  | ! semi_probe_  |
//+------------------------------+--------------+----------------+
//| normal true                  | (right, TRUE)|  (right, ONLY) |
//+------------------------------+--------------+----------------+
//+------------------------------+--------------+----------------+
//|        ! matched             | semi_probe_  | ! semi_probe_  |
//+------------------------------+--------------+----------------+
//|NULL v.s. (empty set)         |              |                |
//|e.g., NULL IN (empty set)     |(right, FALSE)|  NO_OUTPUT     |
//+------------------------------+--------------+----------------+
//|NULL v.s. (set)               |              |                |
//|e.g., NULL IN (1, 2, 3)       |(right, NULL) |  NO_OUTPUT     |
//+------------------------------+--------------+----------------+
//|left_row v.s. (set with NULL) |              |                |
//|e.g., 10 IN (1, NULL, 3)      |(right, NULL) |  NO_OUTPUT     |
//+------------------------------+--------------+----------------+
//|normal false                  |              |                |
//|e.g., 10 IN (1, 2, 3)         |(right, FALSE)|  NO_OUTPUT     |
//+------------------------------+--------------+----------------+

Caso exista um PostFilter, as entradas correspondidas da tabela hash são verificadas pelo PostFilter.

RightAntiSemi

O RIGHT ANTI SEMI JOIN é semelhante ao RIGHT OUTER JOIN. Apenas as tuplas da tabela direita são emitidas, com base na seguinte tabela verdade:

//+------------------------------+----------------+
//|        ! matched             | ! semi_probe_  |
//+------------------------------+----------------+
//|NULL v.s. (empty set)         |                |
//|e.g., NULL NOT IN (empty set) |  (right, ONLY) |
//+------------------------------+----------------+
//|NULL v.s. (set)               |                |
//|e.g., NULL NOT IN (1, 2, 3)   |  (right, ONLY) |
//+------------------------------+----------------+
//|left_row v.s. (set with NULL) |                |
//|e.g., 10 NOT IN (1, NULL, 3)  |  (right, ONLY) |
//+------------------------------+----------------+
//|normal false                  |                |
//|e.g., 10 NOT IN (1, 2, 3)     |  (right, ONLY) |
//+------------------------------+----------------+

Caso exista um PostFilter, as entradas correspondidas da tabela hash são verificadas pelo PostFilter antes da saída.

size_t PutValue(uint64_t hash_code, const char *key_buf, uint64_t key_len, const uint64_t tuple);
ValueIterator FindValue(uint64_t hash_code, const char *key_data, const uint64_t key_len, const bool need_mark = false);

enum IteratorType { Normal = 0, NoneMark = 1, Mark = 2, END };

class TableIterator {
 public:
  void Next();
  bool IsValid() const { return valid_; }
  ValueIterator GetIterator(IteratorType type);

 private:
  IteratorType type_ = IteratorType::END;
};

class ValueIterator {
  struct Listener {
    virtual void BlockEvent() {}
  };
  void SetListener(Listener *listener) { listener_ = listener; }

  void Next();
  bool IsValid() const { return valid_; }

 private:
  IteratorType type_ = IteratorType::Normal;
  Listener *listener_ = nullptr;
};

class HashMatchFetcher final {
  bool Fetch(Context &context, TupleChunk *&mat_chunk);
  // fetch from left or right child
  bool FetchMem(Context &context);
  // fetch from info chunks (include load from temp files)
  bool FetchDisk(Context &context, TupleChunk *&mat_chunk);

  size_t part_index_ = 0;
  TupleChunk chunk_;
};

class HashMatchBuilder {
  void Build();
  virtual void ChunkResult(const size_t offset, const bool is_null,
                           const size_t part_index, const uint64_t hash_val,
                           const char *key_data, const size_t key_len) = 0;
  virtual void ChunkDone() = 0;

  HashMatchFetcher fetcher_;
};

class HashMatchMemBuilder final : public HashMatchBuilder {
  void ChunkResult(const size_t offset, const bool is_null,
                   const size_t part_index, const uint64_t hash_val,
                   const char *key_data, const size_t key_len) override;
  void ChunkDone() override;

  TupleChunk origin_chunk_;
};

class HashMatchDiskBuilder final : public HashMatchBuilder {
  void ChunkResult(const size_t offset, const bool is_null,
                   const size_t part_index, const uint64_t hash_val,
                   const char *key_data, const size_t key_len) override;
  void ChunkDone() override;

  const size_t part_index_ = 0;
};

class HashMatchProber final {
 public:
  void ProbeResult(TupleChunk *tpchunk, size_t &chunk_off, const size_t chunk_size);
  bool ProbeIter(Context &context, TupleChunk *tpchunk, size_t &chunk_off, const size_t chunk_size);
  bool Probe(Context &context, TupleChunk *tpchunk, size_t &chunk_off, const size_t chunk_size, const bool disk);

 private:
  const HashMatch &join_;
  HMInfo *info_ = nullptr;
  HMLocalInfo *local_info_ = nullptr;
  size_t part_index_ = 0;

  PostFilter filter_;

  LeftIterator lit_;
  RightIterator rit_;
  TraverseIterator tit_;  // used for probe left
};

struct PostFilter final {
  bool Evaluate();
  bool Probe(TupleChunk *tpchunk, size_t &chunk_off, const size_t chunk_size);

  const HashMatchProber &prober_;
  const RTExprTreePtr &post_expr_;
  const HashMatchExpr &left_expr_;
  const HashMatchExpr &right_expr_;
  std::shared_ptr<Expressions::ExprEnv> post_env_ = nullptr;
};

struct Iterator {
  virtual void InitExpr() {}
  virtual void FiniExpr() {}
  virtual void Init(const size_t part_index);
  virtual void Fini();
  virtual bool Valid(Context &context) { return false; }
  virtual void Next() = 0;

  HashMatchProber &prober_;
  PostFilter &filter_;
};

// for Probe
struct LeftIterator final : public Iterator, public ValueIterator::Listener {
  void BlockEvent() override;

  bool Valid(Context &context) override;
  void Next() override;
  bool Find(const size_t part_index, const uint64_t hash_val,
            const char *key_data, const uint64_t key_len);

  ValueIterator it_;
};

// for Probe
struct RightIterator : public Iterator {
  bool Valid(Context &context) override;
  void Next() override;

  HashMatchFetcher fetcher_;
  TupleChunk origin_chunk_;
  size_t chunk_size_ = 0;
};

// for ProbeLeft
struct TraverseIterator final : public Iterator {
  bool Valid(Context &context) override;
  void Next() override;

  TableIterator tit_;
  LeftIterator lit_;
  IteratorType it_type_ = IteratorType::END;
};