Bug#30437378 : ASSERTION N_EXT == DTUPLE->GET_N_EXT() AT

               REC_GET_CONVERTED_SIZE() IN REM0REC.IC

Issue:
 n_ext information being passed from row_upd_clust_rec_by_insert() was
 number of ext columns that are part of index. But this information is
 being used as total number of ext columns in further function flow.
 which might differ.

Fix:
 Looking closely at the code, it is seen that rec_get_converted_size()
 doesn't actually need number of ext fields. It just need to know if ext
 fields are there or not.
 - Added a new member function in dfield_t to return true if tuple has
   at least one ext fields.
 - Don't pass n_ext to rec_get_converted_size(). Instead call
   dfield_t->has_ext() to know if ext fields are there or not.
 - Remove the problematic assert as it is not needed now.
 - There were many places where n_ext was being passed to functions but
   it seems none of the places actual number was needed. Only information
   needed is if ext fields exists. Thus, did a complete cleanup of code
   wherever n_ext was being passed.

RB : 23310
Reviewed by : Annamalai Gurusami <[email protected]>
              Jakub Lopuszanski <[email protected]>

Author: mayprasa

storage/innobase/btr/btr0btr.cc

@@ -1490,7 +1490,6 @@ rec_t *btr_root_raise_and_insert(
     ulint **offsets,       /*!< out: offsets on inserted record */
     mem_heap_t **heap,     /*!< in/out: pointer to memory heap, or NULL */
     const dtuple_t *tuple, /*!< in: tuple to insert */
-    ulint n_ext,           /*!< in: number of externally stored columns */
     mtr_t *mtr)            /*!< in: mtr */
 {
   dict_index_t *index;
@@ -1630,8 +1629,8 @@ rec_t *btr_root_raise_and_insert(
 
   page_cur_set_before_first(root_block, page_cursor);
 
-  node_ptr_rec = page_cur_tuple_insert(page_cursor, node_ptr, index, offsets,
-                                       heap, 0, mtr);
+  node_ptr_rec =
+      page_cur_tuple_insert(page_cursor, node_ptr, index, offsets, heap, mtr);
 
   /* The root page should only contain the node pointer
   to new_page at this point.  Thus, the data should fit. */
@@ -1649,11 +1648,11 @@ rec_t *btr_root_raise_and_insert(
   /* Split the child and insert tuple */
   if (dict_index_is_spatial(index)) {
     /* Split rtree page and insert tuple */
-    return (rtr_page_split_and_insert(flags, cursor, offsets, heap, tuple,
-                                      n_ext, mtr));
+    return (
+        rtr_page_split_and_insert(flags, cursor, offsets, heap, tuple, mtr));
   } else {
-    return (btr_page_split_and_insert(flags, cursor, offsets, heap, tuple,
-                                      n_ext, mtr));
+    return (
+        btr_page_split_and_insert(flags, cursor, offsets, heap, tuple, mtr));
   }
 }
 
@@ -1753,8 +1752,7 @@ ibool btr_page_get_split_rec_to_right(
  the lower or upper half-page (determined by btr_page_tuple_smaller()) */
 static rec_t *btr_page_get_split_rec(
     btr_cur_t *cursor,     /*!< in: cursor at which insert should be made */
-    const dtuple_t *tuple, /*!< in: tuple to insert */
-    ulint n_ext)           /*!< in: number of externally stored columns */
+    const dtuple_t *tuple) /*!< in: tuple to insert */
 {
   page_t *page;
   page_zip_des_t *page_zip;
@@ -1773,7 +1771,7 @@ static rec_t *btr_page_get_split_rec(
 
   page = btr_cur_get_page(cursor);
 
-  insert_size = rec_get_converted_size(cursor->index, tuple, n_ext);
+  insert_size = rec_get_converted_size(cursor->index, tuple);
   free_space = page_get_free_space_of_empty(page_is_comp(page));
 
   page_zip = btr_cur_get_page_zip(cursor);
@@ -1871,7 +1869,6 @@ static MY_ATTRIBUTE((warn_unused_result)) bool btr_page_insert_fits(
     ulint **offsets,        /*!< in: rec_get_offsets(
                             split_rec, cursor->index); out: garbage */
     const dtuple_t *tuple,  /*!< in: tuple to insert */
-    ulint n_ext,            /*!< in: number of externally stored columns */
     mem_heap_t **heap)      /*!< in: temporary memory heap */
 {
   page_t *page;
@@ -1887,7 +1884,7 @@ static MY_ATTRIBUTE((warn_unused_result)) bool btr_page_insert_fits(
   ut_ad(!split_rec || !page_is_comp(page) == !rec_offs_comp(*offsets));
   ut_ad(!split_rec || rec_offs_validate(split_rec, cursor->index, *offsets));
 
-  insert_size = rec_get_converted_size(cursor->index, tuple, n_ext);
+  insert_size = rec_get_converted_size(cursor->index, tuple);
   free_space = page_get_free_space_of_empty(page_is_comp(page));
 
   /* free_space is now the free space of a created new page */
@@ -1989,12 +1986,12 @@ void btr_insert_on_non_leaf_level_func(
 
   err = btr_cur_optimistic_insert(
       flags | BTR_NO_LOCKING_FLAG | BTR_KEEP_SYS_FLAG | BTR_NO_UNDO_LOG_FLAG,
-      &cursor, &offsets, &heap, tuple, &rec, &dummy_big_rec, 0, NULL, mtr);
+      &cursor, &offsets, &heap, tuple, &rec, &dummy_big_rec, NULL, mtr);
 
   if (err == DB_FAIL) {
     err = btr_cur_pessimistic_insert(
         flags | BTR_NO_LOCKING_FLAG | BTR_KEEP_SYS_FLAG | BTR_NO_UNDO_LOG_FLAG,
-        &cursor, &offsets, &heap, tuple, &rec, &dummy_big_rec, 0, NULL, mtr);
+        &cursor, &offsets, &heap, tuple, &rec, &dummy_big_rec, NULL, mtr);
     ut_a(err == DB_SUCCESS);
   }
 
@@ -2181,15 +2178,13 @@ of a page.
 @param[out]	offsets	offsets on inserted record
 @param[in,out]	heap	memory heap for allocating offsets
 @param[in]	tuple	tuple to insert
-@param[in]	n_ext	number of externally stored columns
 @param[in,out]	mtr	mini-transaction
 @return	inserted record (first record on the right sibling page);
         the cursor will be positioned on the page infimum
 @retval	NULL if the operation was not performed */
 static rec_t *btr_insert_into_right_sibling(uint32_t flags, btr_cur_t *cursor,
                                             ulint **offsets, mem_heap_t *heap,
-                                            const dtuple_t *tuple, ulint n_ext,
-                                            mtr_t *mtr) {
+                                            const dtuple_t *tuple, mtr_t *mtr) {
   buf_block_t *block = btr_cur_get_block(cursor);
   page_t *page = buf_block_get_frame(block);
   page_no_t next_page_no = btr_page_get_next(page, mtr);
@@ -2234,7 +2229,7 @@ static rec_t *btr_insert_into_right_sibling(uint32_t flags, btr_cur_t *cursor,
   }
 
   rec = page_cur_tuple_insert(&next_page_cursor, tuple, cursor->index, offsets,
-                              &heap, n_ext, mtr);
+                              &heap, mtr);
 
   if (rec == NULL) {
     if (is_leaf && next_block->page.size.is_compressed() &&
@@ -2307,7 +2302,6 @@ rec_t *btr_page_split_and_insert(
     ulint **offsets,       /*!< out: offsets on inserted record */
     mem_heap_t **heap,     /*!< in/out: pointer to memory heap, or NULL */
     const dtuple_t *tuple, /*!< in: tuple to insert */
-    ulint n_ext,           /*!< in: number of externally stored columns */
     mtr_t *mtr)            /*!< in: mtr */
 {
   buf_block_t *block;
@@ -2338,8 +2332,8 @@ rec_t *btr_page_split_and_insert(
 
   if (dict_index_is_spatial(index)) {
     /* Split rtree page and update parent */
-    return (rtr_page_split_and_insert(flags, cursor, offsets, heap, tuple,
-                                      n_ext, mtr));
+    return (
+        rtr_page_split_and_insert(flags, cursor, offsets, heap, tuple, mtr));
   }
 
   if (!*heap) {
@@ -2369,8 +2363,8 @@ rec_t *btr_page_split_and_insert(
   ut_ad(!page_is_empty(page));
 
   /* try to insert to the next page if possible before split */
-  rec = btr_insert_into_right_sibling(flags, cursor, offsets, *heap, tuple,
-                                      n_ext, mtr);
+  rec =
+      btr_insert_into_right_sibling(flags, cursor, offsets, *heap, tuple, mtr);
 
   if (rec != NULL) {
     return (rec);
@@ -2386,7 +2380,7 @@ rec_t *btr_page_split_and_insert(
   if (n_iterations > 0) {
     direction = FSP_UP;
     hint_page_no = page_no + 1;
-    split_rec = btr_page_get_split_rec(cursor, tuple, n_ext);
+    split_rec = btr_page_get_split_rec(cursor, tuple);
 
     if (split_rec == NULL) {
       insert_left =
@@ -2454,10 +2448,10 @@ rec_t *btr_page_split_and_insert(
   insert_empty:
     ut_ad(!split_rec);
     ut_ad(!insert_left);
-    buf = UT_NEW_ARRAY_NOKEY(
-        byte, rec_get_converted_size(cursor->index, tuple, n_ext));
+    buf =
+        UT_NEW_ARRAY_NOKEY(byte, rec_get_converted_size(cursor->index, tuple));
 
-    first_rec = rec_convert_dtuple_to_rec(buf, cursor->index, tuple, n_ext);
+    first_rec = rec_convert_dtuple_to_rec(buf, cursor->index, tuple);
     move_limit = page_rec_get_next(btr_cur_get_rec(cursor));
   }
 
@@ -2474,16 +2468,15 @@ rec_t *btr_page_split_and_insert(
   if (split_rec) {
     insert_will_fit =
         !new_page_zip &&
-        btr_page_insert_fits(cursor, split_rec, offsets, tuple, n_ext, heap);
+        btr_page_insert_fits(cursor, split_rec, offsets, tuple, heap);
   } else {
     if (!insert_left) {
       UT_DELETE_ARRAY(buf);
       buf = NULL;
     }
 
-    insert_will_fit =
-        !new_page_zip &&
-        btr_page_insert_fits(cursor, NULL, offsets, tuple, n_ext, heap);
+    insert_will_fit = !new_page_zip &&
+                      btr_page_insert_fits(cursor, NULL, offsets, tuple, heap);
   }
 
   if (!srv_read_only_mode && !cursor->index->table->is_intrinsic() &&
@@ -2608,7 +2601,7 @@ rec_t *btr_page_split_and_insert(
   page_cur_search(insert_block, cursor->index, tuple, page_cursor);
 
   rec = page_cur_tuple_insert(page_cursor, tuple, cursor->index, offsets, heap,
-                              n_ext, mtr);
+                              mtr);
 
 #ifdef UNIV_ZIP_DEBUG
   {
@@ -2635,7 +2628,7 @@ rec_t *btr_page_split_and_insert(
   }
 
   rec = page_cur_tuple_insert(page_cursor, tuple, cursor->index, offsets, heap,
-                              n_ext, mtr);
+                              mtr);
 
   if (rec == NULL) {
     /* The insert did not fit on the page: loop back to the

storage/innobase/btr/btr0bulk.cc

@@ -167,18 +167,17 @@ dberr_t PageBulk::init() {
 @param[in]  tuple     tuple to insert
 @param[in]  big_rec   external record
 @param[in]  rec_size  record size
-@param[in]  n_ext     number of externally stored columns
 @return error code */
 dberr_t PageBulk::insert(const dtuple_t *tuple, const big_rec_t *big_rec,
-                         ulint rec_size, ulint n_ext) {
+                         ulint rec_size) {
   ulint *offsets = nullptr;
 
   DBUG_EXECUTE_IF("BtrBulk_insert_inject_error", return DB_INTERRUPTED;);
 
   /* Convert tuple to record. */
   byte *rec_mem = static_cast<byte *>(mem_heap_alloc(m_heap, rec_size));
 
-  rec_t *rec = rec_convert_dtuple_to_rec(rec_mem, m_index, tuple, n_ext);
+  rec_t *rec = rec_convert_dtuple_to_rec(rec_mem, m_index, tuple);
   offsets = rec_get_offsets(rec, m_index, offsets, ULINT_UNDEFINED, &m_heap);
 
   /* Insert the record.*/
@@ -891,10 +890,9 @@ dberr_t BtrBulk::prepareSpace(PageBulk *&page_bulk, ulint level,
 @param[in]  big_rec     big record vector, could be nullptr if there is no
                         data to be stored externally.
 @param[in]  rec_size    record size
-@param[in]  n_ext       number of externally stored columns
 @return error code */
 dberr_t BtrBulk::insert(PageBulk *page_bulk, dtuple_t *tuple,
-                        big_rec_t *big_rec, ulint rec_size, ulint n_ext) {
+                        big_rec_t *big_rec, ulint rec_size) {
   dberr_t err = DB_SUCCESS;
 
   if (big_rec != nullptr) {
@@ -910,7 +908,7 @@ dberr_t BtrBulk::insert(PageBulk *page_bulk, dtuple_t *tuple,
     }
   }
 
-  err = page_bulk->insert(tuple, big_rec, rec_size, n_ext);
+  err = page_bulk->insert(tuple, big_rec, rec_size);
 
   if (big_rec != nullptr) {
     /* Restore latches */
@@ -966,19 +964,18 @@ dberr_t BtrBulk::insert(dtuple_t *tuple, ulint level) {
                          dtuple_get_info_bits(tuple) | REC_INFO_MIN_REC_FLAG);
   }
 
-  ulint n_ext = 0;
-  ulint rec_size = rec_get_converted_size(m_index, tuple, n_ext);
+  ulint rec_size = rec_get_converted_size(m_index, tuple);
   big_rec_t *big_rec = nullptr;
 
   if (page_bulk->needExt(tuple, rec_size)) {
     /* The record is so big that we have to store some fields
     externally on separate database pages */
-    big_rec = dtuple_convert_big_rec(m_index, 0, tuple, &n_ext);
+    big_rec = dtuple_convert_big_rec(m_index, 0, tuple);
     if (big_rec == nullptr) {
       return (DB_TOO_BIG_RECORD);
     }
 
-    rec_size = rec_get_converted_size(m_index, tuple, n_ext);
+    rec_size = rec_get_converted_size(m_index, tuple);
   }
 
   if (page_bulk->isTableCompressed() && page_zip_is_too_big(m_index, tuple)) {
@@ -1000,7 +997,7 @@ dberr_t BtrBulk::insert(dtuple_t *tuple, ulint level) {
     }
   });
 
-  err = insert(page_bulk, tuple, big_rec, rec_size, n_ext);
+  err = insert(page_bulk, tuple, big_rec, rec_size);
 
 func_exit:
   if (big_rec != nullptr) {