Bug#37478594 Virtual column value may be incorrectly set to NULL on cascade update

Problem:
When an ON DELETE SET NULL cascade updates a row in a child table,
setting to NULL all the columns on which a virtual column is based,
one of 2 things may happen:
1. the virtual field is also set to NULL in the update, instead of being
re-calculated.
2. The value of the virtual field is not re-calculated at all.
This impacts indexes based on the virtual field, leading to index
corruption.

Root causes:
For both ON DELETE SET NULL and ON UPDATE SET NULL actions the virtual
field calculation attempts to use parent table update vector to detrmine
which base fields should be treated as NULL for the purpose of the
calculation. However, in the ON DELETE SET NULL scenario the parent row
is deleted and consequently there is no meaningful update vector for it.
Additionally, the scenario where the virtual column is based on all the
columns that comprise the FK is treated as special, somewhat hiding the
problem - the virtual field is "calculated" to NULL without actually
calculating the underlying expression. If the expression is such that
its value is NULL if all column operands are NULL, this happens to be
correct - but obviously this not true for all expressions.

Fix:
When virtual fields are calculated as part of a FOREIGN KEY action,
all the updates for the non-virtual columns of the child table are
already known at that point. Instead of looking up updates to the
parent table, these updates may be used directly by the calculation.
This may be applied uniformly to all FK update actions: ON DELETE
SET NULL, ON UPDATE CASCADE and ON UPDATE SET NULL.

Change-Id: I6ee5c0502dc397276edac09513d0e2f27387aeaf

Author: Andrzej Jarzabek

storage/innobase/btr/btr0sea.cc

@@ -417,8 +417,7 @@ static void btr_search_info_update_hash(btr_cur_t *cursor) {
     return;
   }
 
-  const uint16_t n_unique =
-      static_cast<uint16_t>(dict_index_get_n_unique_in_tree(index));
+  const uint16_t n_unique = dict_index_get_n_unique_in_tree(index);
   const auto info = index->search_info;
   if (info->n_hash_potential != 0) {
     const auto prefix_info = info->prefix_info.load();

storage/innobase/data/data0data.cc

@@ -474,7 +474,7 @@ big_rec_t *dtuple_convert_big_rec(dict_index_t *index, upd_t *upd,
     byte *data;
     ulint longest = 0;
     ulint longest_i = ULINT_MAX;
-    upd_field_t *uf = nullptr;
+    const upd_field_t *uf = nullptr;
 
     for (ulint i = dict_index_get_n_unique_in_tree(index);
          i < dtuple_get_n_fields(entry); i++) {
@@ -606,7 +606,8 @@ big_rec_t *dtuple_convert_big_rec(dict_index_t *index, upd_t *upd,
     if (upd == nullptr) {
       big_rec.ext_in_old = false;
     } else {
-      upd_field_t *uf = upd->get_field_by_field_no(longest_i, index);
+      const upd_field_t *const uf =
+          upd->get_field_by_field_no(longest_i, index);
       ut_ad(uf != nullptr);
       big_rec.ext_in_old = uf->ext_in_old;
     }
@@ -657,7 +658,7 @@ big_rec_t *big_rec_t::alloc(mem_heap_t *heap, ulint n_fld) {
       mem_heap_alloc(heap, n_fld * sizeof(big_rec_field_t)));
 
   rec->n_fields = 0;
-  return (rec);
+  return rec;
 }
 
 dfield_t *dfield_t::clone(mem_heap_t *heap) {
@@ -677,13 +678,13 @@ dfield_t *dfield_t::clone(mem_heap_t *heap) {
     obj->data = nullptr;
   }
 
-  return (obj);
+  return obj;
 }
 
 byte *dfield_t::blobref() const {
   ut_ad(ext);
 
-  return (static_cast<byte *>(data) + len - BTR_EXTERN_FIELD_REF_SIZE);
+  return static_cast<byte *>(data) + len - BTR_EXTERN_FIELD_REF_SIZE;
 }
 
 uint32_t dfield_t::lob_version() const {

storage/innobase/ddl/ddl0builder.cc

@@ -793,7 +793,6 @@ dberr_t Builder::get_virtual_column(Copy_ctx &ctx, const dict_field_t *ifield,
                                     size_t &mv_rows_added) noexcept {
   const auto n_added = mv_rows_added;
   auto v_col = reinterpret_cast<const dict_v_col_t *>(col);
-  const auto clust_index = m_ctx.m_new_table->first_index();
   auto key_buffer = m_thread_ctxs[ctx.m_thread_id]->m_key_buffer;
 
   if (col->is_multi_value()) {
@@ -807,8 +806,8 @@ dberr_t Builder::get_virtual_column(Copy_ctx &ctx, const dict_field_t *ifield,
       auto p = m_v_heap.get();
 
       src_field = innobase_get_computed_value(
-          ctx.m_row.m_ptr, v_col, clust_index, &p, key_buffer->heap(), ifield,
-          m_ctx.thd(), ctx.m_my_table, m_ctx.m_old_table, nullptr, nullptr);
+          ctx.m_row.m_ptr, v_col, m_ctx.m_new_table, &p, key_buffer->heap(),
+          m_ctx.thd(), ctx.m_my_table, ifield, m_ctx.m_old_table);
 
       m_v_heap.reset(p);
 
@@ -838,8 +837,8 @@ dberr_t Builder::get_virtual_column(Copy_ctx &ctx, const dict_field_t *ifield,
     auto p = m_v_heap.get();
 
     src_field = innobase_get_computed_value(
-        ctx.m_row.m_ptr, v_col, clust_index, &p, nullptr, ifield, m_ctx.thd(),
-        ctx.m_my_table, m_ctx.m_old_table, nullptr, nullptr);
+        ctx.m_row.m_ptr, v_col, m_ctx.m_new_table, &p, nullptr, m_ctx.thd(),
+        ctx.m_my_table, ifield, m_ctx.m_old_table);
 
     m_v_heap.reset(p);
 

storage/innobase/dict/dict0dict.cc

@@ -2406,9 +2406,9 @@ static void dict_index_try_cache_rec_offsets(dict_index_t *index) {
   This is not an assert crucial for correctness. It's just to show that there's
   no obvious regression w.r.t intrinsic tables. */
   if (index->table->is_intrinsic() && index->n_uniq != n_unique_in_tree) {
-    ut_a(index->n_uniq == n_unique_in_tree - 1);
+    ut_a(index->n_uniq == n_unique_in_tree - 1U);
     ut_a(!index->is_clustered());
-    ut_a(index->get_field(n_unique_in_tree - 1)->fixed_len);
+    ut_a(index->get_field(n_unique_in_tree - 1U)->fixed_len);
   }
 #endif
   for (size_t i = 0; i < n_unique_in_tree; i++) {

storage/innobase/handler/ha_innodb.cc

@@ -23551,53 +23551,32 @@ void innobase_rename_vc_templ(dict_table_t *table) {
   table->vc_templ->tb_name.assign(table_name);
 }
 
-dfield_t *innobase_get_field_from_update_vector(dict_foreign_t *foreign,
-                                                upd_t *update,
-                                                uint32_t col_no) {
-  dict_table_t *parent_table = foreign->referenced_table;
-  dict_index_t *parent_index = foreign->referenced_index;
-  uint32_t parent_field_no;
-  uint32_t parent_col_no;
-  uint32_t child_col_no;
-
-  for (uint32_t i = 0; i < foreign->n_fields; i++) {
-    child_col_no = foreign->foreign_index->get_col_no(i);
-    if (child_col_no != col_no) {
-      continue;
-    }
-    parent_col_no = parent_index->get_col_no(i);
-    parent_field_no = dict_table_get_nth_col_pos(parent_table, parent_col_no);
-    for (uint32_t j = 0; j < update->n_fields; j++) {
-      upd_field_t *parent_ufield = &update->fields[j];
-      if (parent_ufield->field_no == parent_field_no) {
-        return (&parent_ufield->new_val);
-      }
+/** Get the updated field value from an update vector for the given column
+number.
+@param[in]      update   the update vector for the table's clustered index
+@param[in]      col_no   the number of the non-virtual column in the table
+@return the field's new value if it's updated, otherwise nullptr */
+static const dfield_t *innobase_get_field_from_update_vector(
+    const upd_t *update, uint32_t col_no) {
+  const dict_table_t *const table = update->table;
+  const dict_index_t *const clustered_index = table->first_index();
+  const auto index_field_pos = clustered_index->get_col_pos(col_no);
+
+  for (uint32_t j = 0; j < update->n_fields; j++) {
+    const upd_field_t *const ufield = &update->fields[j];
+    if (ufield->field_no == index_field_pos) {
+      return &ufield->new_val;
     }
   }
 
   return (nullptr);
 }
 
-/** Get the computed value by supplying the base column values.
-@param[in,out]  row             the data row
-@param[in]      col             virtual column
-@param[in]      index           index on the virtual column
-@param[in,out]  local_heap      heap memory for processing large data etc.
-@param[in,out]  heap            memory heap that copies the actual index row
-@param[in]      ifield          index field
-@param[in]      thd             MySQL thread handle
-@param[in,out]  mysql_table     mysql table object
-@param[in]      old_table       during ALTER TABLE, this is the old table
-                                or NULL.
-@param[in]      parent_update   update vector for the parent row
-@param[in]      foreign         foreign key information
-@return the field filled with computed value, or NULL if just want
-to store the value in passed in "my_rec" */
 dfield_t *innobase_get_computed_value(
-    const dtuple_t *row, const dict_v_col_t *col, const dict_index_t *index,
-    mem_heap_t **local_heap, mem_heap_t *heap, const dict_field_t *ifield,
-    THD *thd, TABLE *mysql_table, const dict_table_t *old_table,
-    upd_t *parent_update, dict_foreign_t *foreign) {
+    const dtuple_t *row, const dict_v_col_t *col, const dict_table_t *table,
+    mem_heap_t **local_heap, mem_heap_t *heap, THD *thd, TABLE *mysql_table,
+    const dict_field_t *ifield, const dict_table_t *old_table,
+    upd_t *row_update) {
   byte rec_buf1[REC_VERSION_56_MAX_INDEX_COL_LEN];
   byte rec_buf2[REC_VERSION_56_MAX_INDEX_COL_LEN];
   byte *mysql_rec;
@@ -23607,36 +23586,27 @@ dfield_t *innobase_get_computed_value(
   ulong mv_length = 0;
   const char *mv_data_ptr = nullptr;
 
-  const page_size_t page_size = (old_table == nullptr)
-                                    ? dict_table_page_size(index->table)
-                                    : dict_table_page_size(old_table);
-
-  const dict_index_t *clust_index = nullptr;
-  if (old_table == nullptr) {
-    clust_index = index->table->first_index();
-  } else {
-    clust_index = old_table->first_index();
-  }
+  /* table definition to use for externally stored fields */
+  const dict_table_t *const ext_src_table = (old_table ? old_table : table);
+  const page_size_t page_size = dict_table_page_size(ext_src_table);
 
   ulint ret = 0;
 
-  ut_ad(index->table->vc_templ);
+  ut_ad(table->vc_templ);
   ut_ad(thd != nullptr);
 
   const mysql_row_templ_t *vctempl =
-      index->table->vc_templ
-          ->vtempl[index->table->vc_templ->n_col + col->v_pos];
+      table->vc_templ->vtempl[table->vc_templ->n_col + col->v_pos];
 
-  if (!heap ||
-      index->table->vc_templ->rec_len >= REC_VERSION_56_MAX_INDEX_COL_LEN) {
+  if (!heap || table->vc_templ->rec_len >= REC_VERSION_56_MAX_INDEX_COL_LEN) {
     if (*local_heap == nullptr) {
       *local_heap = mem_heap_create(UNIV_PAGE_SIZE, UT_LOCATION_HERE);
     }
 
     mysql_rec = static_cast<byte *>(
-        mem_heap_alloc(*local_heap, index->table->vc_templ->rec_len));
+        mem_heap_alloc(*local_heap, table->vc_templ->rec_len));
     buf = static_cast<byte *>(
-        mem_heap_alloc(*local_heap, index->table->vc_templ->rec_len));
+        mem_heap_alloc(*local_heap, table->vc_templ->rec_len));
   } else {
     mysql_rec = rec_buf1;
     buf = rec_buf2;
@@ -23646,12 +23616,11 @@ dfield_t *innobase_get_computed_value(
     dict_col_t *base_col = col->base_col[i];
     const dfield_t *row_field = nullptr;
     uint32_t col_no = base_col->ind;
-    const mysql_row_templ_t *templ = index->table->vc_templ->vtempl[col_no];
+    const mysql_row_templ_t *const templ = table->vc_templ->vtempl[col_no];
     const byte *data;
 
-    if (parent_update != nullptr) {
-      row_field =
-          innobase_get_field_from_update_vector(foreign, parent_update, col_no);
+    if (row_update != nullptr) {
+      row_field = innobase_get_field_from_update_vector(row_update, col_no);
     }
 
     if (row_field == nullptr) {
@@ -23667,20 +23636,20 @@ dfield_t *innobase_get_computed_value(
       }
 
       data = lob::btr_copy_externally_stored_field(
-          thd_to_trx(thd), clust_index, &len, nullptr, data, page_size,
-          dfield_get_len(row_field), false, *local_heap);
+          thd_to_trx(thd), ext_src_table->first_index(), &len, nullptr, data,
+          page_size, dfield_get_len(row_field), false, *local_heap);
     }
 
     if (len == UNIV_SQL_NULL) {
       mysql_rec[templ->mysql_null_byte_offset] |=
           (byte)templ->mysql_null_bit_mask;
       memcpy(mysql_rec + templ->mysql_col_offset,
-             static_cast<const byte *>(index->table->vc_templ->default_rec +
+             static_cast<const byte *>(table->vc_templ->default_rec +
                                        templ->mysql_col_offset),
              templ->mysql_col_len);
     } else {
       row_sel_field_store_in_mysql_format(
-          mysql_rec + templ->mysql_col_offset, templ, index,
+          mysql_rec + templ->mysql_col_offset, templ, table->first_index(),
           templ->clust_rec_field_no, (const byte *)data, len, ULINT_UNDEFINED);
 
       if (templ->mysql_null_bit_mask) {
@@ -23715,7 +23684,7 @@ dfield_t *innobase_get_computed_value(
         only 1 byte, other BLOBs won't be affected */
         max_len = 255;
       } else {
-        max_len = DICT_MAX_FIELD_LEN_BY_FORMAT(index->table) + 1;
+        max_len = DICT_MAX_FIELD_LEN_BY_FORMAT(table) + 1;
       }
 
       byte *blob_mem = static_cast<byte *>(mem_heap_alloc(heap, max_len));
@@ -23725,8 +23694,8 @@ dfield_t *innobase_get_computed_value(
     }
 
     /* open a temporary table handle */
-    mysql_table = tblhdl.open(thd, index->table->vc_templ->db_name.c_str(),
-                              index->table->vc_templ->tb_name.c_str());
+    mysql_table = tblhdl.open(thd, table->vc_templ->db_name.c_str(),
+                              table->vc_templ->tb_name.c_str());
   }
   if (mysql_table) {
     ret = handler::my_eval_gcolumn_expr(
@@ -23763,8 +23732,8 @@ dfield_t *innobase_get_computed_value(
     json_binary::Value v(json_binary::parse_binary(mv_data_ptr, mv_length));
     multi_value_data *value = nullptr;
 
-    bool succ = innobase_store_multi_value(
-        v, value, fld, field, dict_table_is_comp(index->table), heap);
+    bool succ = innobase_store_multi_value(v, value, fld, field,
+                                           dict_table_is_comp(table), heap);
     if (!succ) {
       ut_error;
     }
@@ -23773,7 +23742,7 @@ dfield_t *innobase_get_computed_value(
   } else {
     row_mysql_store_col_in_innobase_format(
         field, buf, true, mysql_rec + vctempl->mysql_col_offset,
-        vctempl->mysql_col_len, dict_table_is_comp(index->table));
+        vctempl->mysql_col_len, dict_table_is_comp(table));
   }
   field->type.prtype |= DATA_VIRTUAL;
 

storage/innobase/include/data0data.h

@@ -157,7 +157,7 @@ If len>0, tests the first len bytes of the content for equality.
 @return true if both fields are NULL or if they are equal */
 [[nodiscard]] inline bool dfield_datas_are_binary_equal(const dfield_t *field1,
                                                         const dfield_t *field2,
-                                                        ulint len);
+                                                        ulint len = 0);
 /** Tests if dfield data length and content is equal to the given.
 @param[in] field Field
 @param[in] len   Data length or UNIV_SQL_NULL

storage/innobase/include/dict0dict.h

@@ -693,7 +693,7 @@ added column.
 /** Gets the number of fields in the internal representation of an index,
  including fields added by the dictionary system.
  @return number of fields */
-[[nodiscard]] static inline ulint dict_index_get_n_fields(
+[[nodiscard]] inline uint16_t dict_index_get_n_fields(
     const dict_index_t *index); /*!< in: an internal
                                representation of index (in
                                the dictionary cache) */
@@ -702,14 +702,14 @@ added column.
  we do not take multiversioning into account: in the B-tree use the value
  returned by dict_index_get_n_unique_in_tree.
  @return number of fields */
-[[nodiscard]] static inline ulint dict_index_get_n_unique(
+[[nodiscard]] inline uint16_t dict_index_get_n_unique(
     const dict_index_t *index); /*!< in: an internal representation
                                of index (in the dictionary cache) */
 /** Gets the number of fields in the internal representation of an index
  which uniquely determine the position of an index entry in the index, if
  we also take multiversioning into account.
  @return number of fields */
-[[nodiscard]] static inline ulint dict_index_get_n_unique_in_tree(
+[[nodiscard]] inline uint16_t dict_index_get_n_unique_in_tree(
     const dict_index_t *index); /*!< in: an internal representation
                                of index (in the dictionary cache) */
 
@@ -723,14 +723,14 @@ index, if we also take multiversioning into account. Note, it doesn't
 include page no field.
 @param[in]      index   index
 @return number of fields */
-[[nodiscard]] static inline uint16_t dict_index_get_n_unique_in_tree_nonleaf(
+[[nodiscard]] inline uint16_t dict_index_get_n_unique_in_tree_nonleaf(
     const dict_index_t *index);
 /** Gets the number of user-defined ordering fields in the index. In the
  internal representation we add the row id to the ordering fields to make all
  indexes unique, but this function returns the number of fields the user defined
  in the index as ordering fields.
  @return number of fields */
-[[nodiscard]] static inline ulint dict_index_get_n_ordering_defined_by_user(
+[[nodiscard]] inline ulint dict_index_get_n_ordering_defined_by_user(
     const dict_index_t *index); /*!< in: an internal representation
                                of index (in the dictionary cache) */
 /** Returns true if the index contains a column or a prefix of that column.