Address Review comments

Author: satya-bodapati

storage/innobase/btr/btr0btr.cc

@@ -4103,7 +4103,8 @@ static bool btr_validate_level(
     dict_index_t *index, /*!< in: index tree */
     const trx_t *trx,    /*!< in: transaction or NULL */
     ulint level,         /*!< in: level number */
-    bool lockout)        /*!< in: true if X-latch index is intended */
+    bool lockout,        /*!< in: true if X-latch index is intended */
+    blob_ref_map *blob_map = nullptr) /*!< in: Optional blob reference map. */
 {
   buf_block_t *block;
   page_t *page;
@@ -4260,7 +4261,7 @@ static bool btr_validate_level(
 
     ret = false;
 
-  } else if (!page_validate(page, index)) {
+  } else if (!page_validate(page, index, blob_map)) {
     btr_validate_report1(index, level, block);
     ret = false;
 
@@ -4573,9 +4574,10 @@ static bool btr_validate_spatial_index(
 /** Checks the consistency of an index tree.
  @return true if ok */
 bool btr_validate_index(
-    dict_index_t *index, /*!< in: index */
-    const trx_t *trx,    /*!< in: transaction or NULL */
-    bool lockout)        /*!< in: true if X-latch index is intended */
+    dict_index_t *index,    /*!< in: index */
+    const trx_t *trx,       /*!< in: transaction or NULL */
+    bool lockout,           /*!< in: true if X-latch index is intended */
+    blob_ref_map *blob_map) /*!< in: Optional blob reference map. */
 {
   /* Full Text index are implemented by auxiliary tables,
   not the B-tree */
@@ -4640,7 +4642,7 @@ bool btr_validate_index(
   ulint n = btr_page_get_level(root);
 
   for (ulint i = 0; i <= n; ++i) {
-    if (!btr_validate_level(index, trx, n - i, lockout)) {
+    if (!btr_validate_level(index, trx, n - i, lockout, blob_map)) {
       ok = false;
       break;
     }

storage/innobase/handler/ha_innodb.cc

@@ -18916,18 +18916,16 @@ int ha_innobase::check(THD *thd,                /*!< in: user thread handle */
       CHECK TABLE. */
       srv_fatal_semaphore_wait_extend.fetch_add(1);
 
+      blob_ref_map blob_map;
+      blob_ref_map *blob_map_ptr = nullptr;
+
       if (is_extended && index->is_clustered()) {
-        // Setup the thread local map for clustered index only
-        thread_local_blob_map = new blob_ref_map();
+        // Setup the blob map for clustered index only.
+        blob_map_ptr = &blob_map;
       }
 
-      auto blob_ref_clear_guard = create_scope_guard([]() {
-        if (!thread_local_blob_map) return;
-        delete thread_local_blob_map;
-        thread_local_blob_map = nullptr;
-      });
-
-      bool valid = btr_validate_index(index, m_prebuilt->trx, false);
+      bool valid =
+          btr_validate_index(index, m_prebuilt->trx, false, blob_map_ptr);
 
       /* Restore the fatal lock wait timeout after
       CHECK TABLE. */

storage/innobase/include/btr0btr.h

@@ -41,6 +41,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include "gis0type.h"
 #include "mtr0mtr.h"
 #include "page0cur.h"
+#include "page0page.h"
 #include "univ.i"
 
 /** Maximum record size which can be stored on a page, without using the
@@ -553,7 +554,8 @@ the index.
 [[nodiscard]] bool btr_validate_index(
     dict_index_t *index, /*!< in: index */
     const trx_t *trx,    /*!< in: transaction or 0 */
-    bool lockout);       /*!< in: true if X-latch index is intended */
+    bool lockout,        /*!< in: true if X-latch index is intended */
+    blob_ref_map *blob_map = nullptr); /*!< in: Optional blob reference map. */
 
 /** Creates SDI index and stores the root page numbers in page 1 & 2
 @param[in]      space_id        tablespace id

storage/innobase/include/page0page.h

@@ -761,12 +761,22 @@ bool page_simple_validate_old(
 bool page_simple_validate_new(
     const page_t *page); /*!< in: index page in ROW_FORMAT!=REDUNDANT */
 
+/** A blob map to track the first page no of external LOB and its parent record
+which is the <page_no, heap_no>. This is used to find duplicate external LOB
+pages that is shared between two records. This can happen only on corruption
+(cause unknown yet). CHECK TABLE  t1 EXTENDED will use this map to report
+corruption and mark the table as corrupted */
+using blob_ref_map = std::unordered_map<page_no_t, std::pair<page_no_t, ulint>>;
+
 /** This function checks the consistency of an index page.
 @param[in]  page   index page
 @param[in]  index  data dictionary index containing the page record type
 definition
+@param[in]  blob_map Optional blob reference map for tracking, nullptr by
+default.
 @return true if ok */
-bool page_validate(const page_t *page, dict_index_t *index);
+bool page_validate(const page_t *page, dict_index_t *index,
+                   blob_ref_map *blob_map = nullptr);
 
 /** Looks in the page record list for a record with the given heap number.
  @return record, NULL if not found */
@@ -802,23 +812,18 @@ param[in]       index   index
 @return true if ok */
 bool page_is_spatial_non_leaf(const rec_t *rec, dict_index_t *index);
 
-/** A blob map to track the first page no of external LOB and its parent record
-which is the <page_no, heap_no>. This is used to find duplicate external LOB
-pages that is shared between two records. This can happen only on corruption
-(cause unknown yet). CHECK TABLE  t1 EXTENDED will use this map to report
-corruption and mark the table as corrupted */
-using blob_ref_map = std::unordered_map<page_no_t, std::pair<page_no_t, ulint>>;
-extern thread_local blob_ref_map *thread_local_blob_map;
-
 /** Validate that the external LOB's first page is not shared between records of
 a clustered index
 @param[in] rec     physical record
 @param[in] index   index of the table
 @param[in] offsets the record offset array
+@param[in] blob_map Optional blob reference map for tracking, nullptr by
+default.
 @return true If OK else false if external LOB is found to be shared between two
 records, ie false on failure */
 bool page_rec_blob_validate(const rec_t *rec, const dict_index_t *index,
-                            const ulint *offsets);
+                            const ulint *offsets,
+                            blob_ref_map *blob_map = nullptr);
 
 #include "page0page.ic"
 

storage/innobase/page/page0page.cc

@@ -48,12 +48,6 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #endif /* !UNIV_HOTBACKUP */
 #include "lob0lob.h"
 
-/** A blob map to track the first page no of external LOB and its parent record
-which is the <page_no, heap_no>. This is used to find duplicate external LOB
-pages that is shared between two records. This can happen only on corruption
-(cause unknown yet). CHECK TABLE  t1 EXTENDED will use this map to report
-corruption and mark the table as corrupted */
-thread_local blob_ref_map *thread_local_blob_map = nullptr;
 
 /*                      THE INDEX PAGE
                         ==============
@@ -1734,47 +1728,49 @@ a clustered index
 @param[in] rec     physical record
 @param[in] index   index of the table
 @param[in] offsets the record offset array
+@param[in] blob_map Optional blob reference map for tracking, nullptr by
+default.
 @return true If OK else false if external LOB is found to be shared between two
 records, ie false on failure */
 bool page_rec_blob_validate(const rec_t *rec, const dict_index_t *index,
-                            const ulint *offsets) {
-  // this means reference check is not enabled. Enabled only via
-  // CHECK TABLE path
-  if (thread_local_blob_map == nullptr) {
+                            const ulint *offsets, blob_ref_map *blob_map) {
+  // This means reference check is not enabled. Enabled only via
+  // CHECK TABLE path.
+  if (blob_map == nullptr) {
     return true;
   }
 
-  // if index is not PRIMARY, return true
+  // If index is not PRIMARY, return true.
   if (!index->is_clustered()) {
     return true;
   }
 
-  // if page-level is not zero, return true because blob exists only on leaf
-  // level
+  // If page-level is not zero, return true because blob exists only on leaf
+  // level.
   const page_t *page = page_align(rec);
   if (!page_is_leaf(page)) {
     return true;
   }
 
-  // if rec is not user record, blobs dont exist, return true
+  // If rec is not user record, blobs dont exist, return true.
   if (!page_rec_is_user_rec(rec)) {
     return true;
   }
 
-  // if rec doesn't have any external LOB, return true
+  // If rec doesn't have any external LOB, return true.
   if (!rec_offs_any_extern(offsets)) {
     return true;
   }
 
-  // if rec is deleted marked, return true, we cannot validate the blob. the
-  // blob pages in the deleted marked records could be freed
+  // If rec is deleted marked, return true, we cannot validate the blob. The
+  // blob pages in the deleted marked records could be freed.
   if (rec_get_deleted_flag(rec, rec_offs_comp(offsets))) {
     return true;
   }
 
-  // if rec is not the owner of the blob, we cannot validate if blob page state
+  // If rec is not the owner of the blob, we cannot validate if blob page state
   // now validate that the blob first page is not marked as free from page
-  // bitmap
+  // bitmap.
 
   ulint n_fields = rec_offs_n_fields(offsets);
 
@@ -1786,8 +1782,7 @@ bool page_rec_blob_validate(const rec_t *rec, const dict_index_t *index,
           lob::btr_rec_get_field_ref(index, rec, offsets, i));
 
       lob::ref_t ref(field_ref);
-      if (!ref.is_owner() || ref.is_null() || ref.is_null_relaxed() ||
-          ref.is_being_modified()) {
+      if (!ref.is_owner() || ref.is_null_relaxed() || ref.is_being_modified()) {
         continue;
       }
 
@@ -1816,16 +1811,16 @@ bool page_rec_blob_validate(const rec_t *rec, const dict_index_t *index,
 
       DBUG_EXECUTE_IF(
           "simulate_lob_corruption",
-          // introduce corruption after 5 external LOB entries
-          if (thread_local_blob_map->size() >= 5) {
-            // we introduce a fake entry in the map
-            (*thread_local_blob_map)[blob_page_no] = std::make_pair(
+          // Introduce corruption after 5 external LOB entries.
+          if (blob_map->size() >= 5) {
+            // We introduce a fake entry in the map.
+            (*blob_map)[blob_page_no] = std::make_pair(
                 page_get_page_no(page) - 1, page_rec_get_heap_no(rec) - 1);
           });
 
-      auto it = thread_local_blob_map->find(blob_page_no);
-      if (it == thread_local_blob_map->end()) {
-        (*thread_local_blob_map)[blob_page_no] =
+      auto it = blob_map->find(blob_page_no);
+      if (it == blob_map->end()) {
+        (*blob_map)[blob_page_no] =
             std::make_pair(page_get_page_no(page), page_rec_get_heap_no(rec));
       } else {
         auto val = it->second;
@@ -2249,7 +2244,8 @@ bool page_is_spatial_non_leaf(const rec_t *rec, dict_index_t *index) {
   return (dict_index_is_spatial(index) && !page_is_leaf(page_align(rec)));
 }
 
-bool page_validate(const page_t *page, dict_index_t *index) {
+bool page_validate(const page_t *page, dict_index_t *index,
+                   blob_ref_map *blob_map) {
   const page_dir_slot_t *slot;
   mem_heap_t *heap;
   byte *buf;
@@ -2360,7 +2356,8 @@ bool page_validate(const page_t *page, dict_index_t *index) {
       goto func_exit;
     }
 
-    if (!page_rec_blob_validate(const_cast<byte *>(rec), index, offsets)) {
+    if (!page_rec_blob_validate(const_cast<byte *>(rec), index, offsets,
+                                blob_map)) {
       goto func_exit;
     }