MySQL 8.0.22執行器源碼分析HashJoin —— 一些初始化函數的細節步驟

目錄
InitRowBuffer（101行~126行）
InitProbeIterator（142行~153行）
*HashJoinIterator* 的Init（155行~240行）
InitializeChunkFiles（364行~401行）
InitWritingToProbeRowSavingFile（1115~1119）
InitReadingFromProbeRowSavingFile（1121~1126）

讓我們接著上一篇分析BuildHashTable函數細節步驟繼續吧,這些函數都在hash_join_iterator.cc文件中。

InitRowBuffer（101行~126行）

需要注意的兩個步驟

1、初始化row buffer，使用到一個seed。具體哈希計算算法我并沒有特定去了解，這里也就不深入了

kHashTableSeed是個沒有意義的數字，是用于計算hash的key。一個種子在內存哈希表中進行哈希運算，一個種子計算用于確定chunk 文件應該放置的行。如果兩個操作使用同一個種子，將會把塊文件加載到哈希表中，這樣是錯誤的。

2、初始化row buffer后，需要調整迭代器的指向，將其頭尾均指向row buffer的尾部。

......
if (m_row_buffer.Init(kHashTableSeed)) {DBUG_ASSERT(thd()->is_error());  // my_error should have been called.return true;
}m_hash_map_iterator = m_row_buffer.end();
m_hash_map_end = m_row_buffer.end();
return false;

InitProbeIterator（142行~153行）

1、初始化m_probe_input迭代器（為一個RowIterator）

2、判斷m_probe_input_batch_mode是否為真（默認為false，指的是不開啟批處理模式），為真的話就調用StartPSIBatchMode

  if (m_probe_input->Init()) {return true;}if (m_probe_input_batch_mode) {m_probe_input->StartPSIBatchMode();}return false;

HashJoinIterator 的Init（155行~240行）

1、準備從build（驅動表）輸入中讀取行數據到哈希表中，用于構建哈希表

  PrepareForRequestRowId(m_build_input_tables.tables(),m_tables_to_get_rowid_for);

函數調用棧：

|PrepareForRequestRowId
||prepare_for_position

涉及到使用主鍵去找rows，必須用主鍵字段擴展讀取位圖

然后初始化build的迭代器（為一個RowIterator）

2、初始化一些亂七八糟的變量

//默認在內存中做所有操作，并且沒有任何對哈希表的重新填充操作。每個輸入都只讀一次，不會對磁盤寫入任何數據。 
m_hash_join_type = HashJoinType::IN_MEMORY;		
//不需要把被驅動表讀出來的行數據寫到saving files中。因為只有當哈希連接生成塊不能完全放到內存中 或者 哈希連接不能延伸到磁盤時，即probe輸入行需要多次Read才需要開啟probe行保存
m_write_to_probe_row_saving = false;
//很顯然此時build迭代器讀取的buffer中是有行數據的，當build input數據被消耗掉，停止hash join 迭代器請求更多行
m_build_iterator_has_more_rows = true;
//開啟批處理模式
m_probe_input->EndPSIBatchModeIfStarted();
//如過外連接溢出到磁盤上操作，那么probe行可以和我們未看到的build input中的row匹配，若匹配為true。這里默認是內存里操作，所以初始化為false
m_probe_row_match_flag = false;

3、計算build row、probe row 占的內存大小

計算給定表單行數據需要占多大的byte，記錄上界，這樣之后pack的數據長度總是會比這個長度短。

upper_row_size 為build row、probe row兩者的最大值。

  size_t upper_row_size = 0;if (!m_build_input_tables.has_blob_column()) {upper_row_size =hash_join_buffer::ComputeRowSizeUpperBound(m_build_input_tables);}if (!m_probe_input_tables.has_blob_column()) {upper_row_size = std::max(upper_row_size,hash_join_buffer::ComputeRowSizeUpperBound(m_probe_input_tables));}

4、一個看不懂的操作，將一個string類型的變量reverse了一下

if (m_temporary_row_and_join_key_buffer.reserve(upper_row_size)) {my_error(ER_OUTOFMEMORY, MYF(0), upper_row_size);return true;  // oom
}

5、如果一個表包含一個geometry列，確保該數據復制到行緩沖區，而不是只設置指向數據的指針。因為當hash join溢出到磁盤上，需要從塊文件讀回一行，行數據存儲在一個臨時緩沖區中，當臨時緩沖區用于其他用途時，字段指向的數據將變為無效。

  MarkCopyBlobsIfTableContainsGeometry(m_probe_input_tables);MarkCopyBlobsIfTableContainsGeometry(m_build_input_tables);

6、chunk 數組、標志clear操作

//m_chunk_files_on_disk數組來保存磁盤上的塊文件列表，以防我們降級為磁盤上的散列聯接.此時需要clear
m_chunk_files_on_disk.clear();
//build 和 probe 當前從hash join chunk中讀取的是第0行
m_build_chunk_current_row = 0;
m_probe_chunk_current_row = 0;
//現在不使用任何一種hash join chunk
m_current_chunk = -1;

7、對于每個給定的表，如果需要，請求填寫行ID（相當于調用file->position（））

 PrepareForRequestRowId(m_probe_input_tables.tables(),m_tables_to_get_rowid_for);

8、構建哈希表

// Build the hash table
if (BuildHashTable()) {DBUG_ASSERT(thd()->is_error() ||thd()->killed);  // my_error should have been called.return true;
}

9、當build 和 probe 都沒有row數據了，返回。

  if (m_state == State::END_OF_ROWS) {// BuildHashTable() decided that the join is done (the build input is// empty, and we are in an inner-/semijoin. Anti-/outer join must output// NULL-complemented rows from the probe input).return false;}

10、如果是反連接并且join情況數組為空并且沒有其他情況并且此時row buffer中仍然有數據。說明此時不需要輸出任何東西，因為所有數據都在哈希表中。

  if (m_join_type == JoinType::ANTI && m_join_conditions.empty() &&m_extra_condition == nullptr && !m_row_buffer.empty()) {// For degenerate antijoins, we know we will never output anything// if there's anything in the hash table, so we can end right away.// (We also don't need to read more than one row, but// CreateHashJoinAccessPath() has already added a LIMIT 1 for us// in this case.)m_state = State::END_OF_ROWS;return false;}

11、初始化probe迭代器

return InitProbeIterator();

InitializeChunkFiles（364行~401行）

初始化兩個input的hashjoinchunk。

先估計需要多少塊，有一個來源是planner估計的。此外可以假設當前行緩沖區代表了總體的行密度，將估計的剩余row數量/目前讀到的row數量，就能得到chunk數。

1、縮減后的哈希表中的行 = 縮減因子* 哈希表中行數這是一種保護措施，因為我們寧愿得到一個或兩個額外的塊，而不必多次重新讀取探測輸入。

constexpr double kReductionFactor = 0.9;
const size_t reduced_rows_in_hash_table =std::max<size_t>(1, rows_in_hash_table * kReductionFactor)；

2、剩余行數 = max(哈希表中行數,planner估計的join的行數) - 哈希表中的行數

const size_t remaining_rows =std::max(rows_in_hash_table, estimated_rows_produced_by_join) -rows_in_hash_table;

3、需要的chunk數 = max(剩余行數 / 縮減后的哈希表中的行數)

 const size_t chunks_needed = std::max<size_t>(1, std::ceil(remaining_rows / reduced_rows_in_hash_table));

4、真正的chunk數 = min(最大chunk文件數，需要的chunk數) 限制每個輸入的塊數，這樣就不會冒著達到服務器對打開文件數的限制的風險。

const size_t num_chunks = std::min(max_chunk_files, chunks_needed);

5、確保chunk數目是偶數，因為我們join的時候是按照probe和build的一對chunk進行join的

  const size_t num_chunks_pow_2 = my_round_up_to_next_power(num_chunks);

6、調整chunk數目到偶數；然后對每對chunk進行初始化，一個是buildchunk一個是probechunk

  chunk_pairs->resize(num_chunks_pow_2);for (ChunkPair &chunk_pair : *chunk_pairs) {if (chunk_pair.build_chunk.Init(build_tables, /*uses_match_flags=*/false) ||chunk_pair.probe_chunk.Init(probe_tables,include_match_flag_for_probe)) {my_error(ER_TEMP_FILE_WRITE_FAILURE, MYF(0));return true;}}

對于Init的兩個參數解釋：

tables – 行數據存儲在哪個input
uses_match_flags – 是否應在每行前面加上匹配標志，表示該行是否有匹配行。

InitWritingToProbeRowSavingFile（1115~1119）

標記已經啟用probe row saving，并準備probe row saving file以供寫入

m_write_to_probe_row_saving = true;return m_probe_row_saving_write_file.Init(m_probe_input_tables,m_join_type == JoinType::OUTER);

至于HashJoinChunk::Init函數我們就不去細究了，它涉及到了IOcache操作。

InitReadingFromProbeRowSavingFile（1121~1126）

標記我們從probe row saving files中讀取數據。然后將saving files 倒回開頭

m_probe_row_saving_read_file = std::move(m_probe_row_saving_write_file);
m_probe_row_saving_read_file_current_row = 0;
m_read_from_probe_row_saving = true;
return m_probe_row_saving_read_file.Rewind();