:Utilisation de Hash Match pour les opérations IMCI

doBuild

Chaque thread worker construit des tables de hachage indépendantes pour chaque partition à partir des données de la table de gauche, produisant ainsi une matrice worker x partition de tables de hachage :

Chaque entrée de la table de hachage est une valeur UInt64 qui encode la position du chunk correspondant à cette clé. La valeur UInt64 se divise en trois champs au niveau binaire :

• UInt16 : le thread worker propriétaire du chunk

• UInt16 : le décalage du chunk

• UInt32 : l'index du tableau au sein du chunk

Les threads workers récupèrent les tuples de la table de gauche en parallèle et les insèrent dans la table de hachage assignée à leur worker et à leur partition, sans verrouillage. Ce processus se poursuit jusqu'au traitement complet des données de la table de gauche.

doMerge

Après l'exécution de doBuild, chaque partition possède une table de hachage par thread worker. La phase de sondage nécessitant une seule table de hachage par partition, doMerge consolide toutes les tables de hachage par worker d'une partition en une seule :

Un seul thread worker fusionne chaque partition de manière indépendante, ce qui élimine la synchronisation par verrous. Comme le nombre de partitions dépasse celui des threads workers, la charge de travail se répartit uniformément entre les workers.

Déversement sur disque lors de la phase de construction

Chaque thread worker reçoit une part égale du quota mémoire total. Lorsque la mémoire disponible ne peut pas contenir toutes les tables de hachage et tous les chunks, l'opérateur déverse des partitions entières sur le disque, en commençant par la partition au numéro le plus élevé, jusqu'à ce que les partitions restantes en mémoire puissent être traitées normalement. Si aucune partition ne tient en mémoire, une erreur OutOfMemoryError est levée.

Pendant l'exécution de doBuild, si un thread worker ne peut pas ajouter une paire clé-valeur à une table de hachage ou stocker un chunk en raison d'un manque de mémoire, il déverse les données de sa partition au numéro le plus élevé sur le disque :

1. Les chunks sont écrits dans un fichier temporaire et leur mémoire est libérée.

2. La table de hachage de cette partition est supprimée.

3. Lorsque la partition devient nécessaire ultérieurement, les chunks sont lus depuis le fichier temporaire et la table de hachage est reconstruite.

Tous les threads workers voient le numéro de la partition déversée la plus élevée. Si la partition déversée la plus élevée d'un worker dépasse le maximum global actuel, il met à jour la valeur globale et libère la mémoire en conséquence. Pour les partitions dont le numéro est supérieur à celui de la partition en mémoire la plus élevée actuelle, les threads workers ne construisent pas de tables de hachage ; ils écrivent plutôt les données directement dans des chunks, qu'ils déversent sur le disque une fois pleins.

doFetch

Les threads workers récupèrent les données de la table de droite et recherchent les tuples correspondants dans la table de hachage de la partition de chaque ligne. Les tuples trouvés sont traités selon le type de jointure. Ce cycle se répète jusqu'à ce que tous les threads workers aient consommé la table de droite.

Déversement sur disque lors de la phase de sondage

Lorsque la phase de construction déverse certaines partitions sur le disque, la phase de sondage traite séparément les partitions en mémoire et celles sur disque.

Pendant l'exécution de doFetch, après qu'un thread worker a récupéré un tuple de la table de droite :

• Si la partition du tuple est en mémoire, le worker effectue une recherche directe dans la table de hachage.

• Si la partition est sur disque, le tuple est stocké dans le chunk de cette partition. Une fois le chunk plein, il est vidé sur le disque et la mémoire du chunk est libérée.

Une fois que tous les workers ont terminé la lecture de la table de droite, les partitions en mémoire sont traitées et leur mémoire est libérée. Les partitions sur disque sont ensuite traitées une par une, chacune étant gérée indépendamment par un seul thread worker (aucun verrouillage requis). Étant donné que le nombre de partitions dépasse celui des threads workers, la charge de travail reste équilibrée.

Le traitement de chaque partition sur disque suit le même schéma construction-sondage :

1. Construction : Les threads workers lisent les données de partition de la table de gauche depuis les fichiers temporaires, désérialisent les chunks et construisent la table de hachage.

2. Sondage : Les threads workers lisent les données de partition de la table de droite depuis les fichiers temporaires, désérialisent les chunks et recherchent les correspondances.

Une fois toutes les partitions traitées, l'opérateur Hash Match se termine. Bien que le traitement des partitions en mémoire et sur disque soit décrit séparément ici, les deux chemins partagent une implémentation de code unique et unifiée.

Étapes du processus de sondage

Toutes les opérations de sondage passent par trois étapes, chacune exécutée par la fonction probe :

1. ProbeMem : Lit les données de la table de droite et dirige chaque tuple soit vers un traitement en mémoire (recherche dans la table de hachage), soit vers un stockage sur disque (écriture dans un fichier temporaire pour un traitement ultérieur dans ProbeDisk).

2. ProbeLeft : Gère les variantes de LEFT JOIN (LEFT OUTER JOIN, LEFT SEMI JOIN, LEFT ANTI SEMI JOIN). Parcourt toutes les paires clé-valeur de la table de hachage et filtre les tuples correspondants ou non correspondants.

3. ProbeDisk : Traite les partitions sur disque une par une en chargeant les chunks depuis les fichiers temporaires et en exécutant la fonction probe. Pour les opérations LEFT JOIN, la logique de jointure s'applique également à chaque partition sur disque.

Inner

Lorsqu'un tuple de la table de droite est non NULL et correspond à la table de hachage construite à partir de la table de gauche, les tuples de gauche et de droite sont émis.

LeftOuter

Lorsqu'un tuple de la table de droite est non NULL et correspond, les tuples de gauche et de droite sont émis. Les tuples de la table de gauche sans correspondance sont émis avec des valeurs NULL aux positions de la table de droite. Si un PostFilter est présent, le PostFilter vérifie les entrées correspondantes de la table de hachage avant l'émission.

RightOuter

Lorsqu'un tuple de la table de droite est non NULL et correspond, les tuples de gauche et de droite sont émis. Les tuples de la table de droite sans correspondance sont émis avec des valeurs NULL aux positions de la table de gauche. Si un PostFilter est présent, le PostFilter vérifie les entrées correspondantes de la table de hachage avant l'émission.

LeftSemi

LEFT SEMI JOIN suit une structure similaire à LEFT OUTER JOIN, mais seuls les tuples de la table de gauche sont émis, et non ceux de la table de droite. La sortie dépend des résultats de correspondance et de l'état de semi_probe_, conformément à cette table de vérité :

//+------------------------------+--------------+----------------+
//|           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     |
//+------------------------------+--------------+----------------+

Si un PostFilter est présent, le PostFilter vérifie les entrées correspondantes de la table de hachage.

LeftAntiSemi

LEFT ANTI SEMI JOIN est similaire à LEFT OUTER JOIN. Seuls les tuples de la table de gauche sont émis, selon la table de vérité suivante :

//+------------------------------+----------------+
//|        ! 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) |
//+------------------------------+----------------+

Si un PostFilter est présent, le PostFilter vérifie les entrées correspondantes de la table de hachage.

RightSemi

RIGHT SEMI JOIN est similaire à RIGHT OUTER JOIN. Seuls les tuples de la table de droite sont émis, selon cette table de vérité :

//+------------------------------+--------------+----------------+
//|           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     |
//+------------------------------+--------------+----------------+

Si un PostFilter est présent, le PostFilter vérifie les entrées correspondantes de la table de hachage.

RightAntiSemi

RIGHT ANTI SEMI JOIN est similaire à RIGHT OUTER JOIN. Seuls les tuples de la table de droite sont émis, selon la table de vérité suivante :

//+------------------------------+----------------+
//|        ! 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) |
//+------------------------------+----------------+

Si un PostFilter est présent, le PostFilter vérifie les entrées correspondantes de la table de hachage avant l'émission.

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;
};