MDEV-24620 ASAN heap-buffer-overflow in btr_pcur_restore_position()

Between btr_pcur_store_position() and btr_pcur_restore_position() it is possible that purge empties a table and enlarges index->n_core_fields and index->n_core_null_bytes. Therefore, we must cache index->n_core_fields in btr_pcur_t::old_n_core_fields so that btr_pcur_t::old_rec can be parsed correctly. Unfortunately, this is a huge change, because we will replace "bool leaf" parameters with "ulint n_core" (passing index->n_core_fields, or 0 for non-leaf pages). For special cases where we know that index->is_instant() cannot hold, we may also pass index->n_fields.

Author: dr-m

mysql-test/suite/innodb/r/instant_alter_debug.result

@@ -317,5 +317,33 @@ SELECT * FROM t1 WHERE c<>1 ORDER BY c DESC;
 c	d
 DROP TABLE t1;
 SET GLOBAL innodb_limit_optimistic_insert_debug = @saved_limit;
+#
+# MDEV-24620 ASAN heap-buffer-overflow in btr_pcur_restore_position()
+#
+CREATE TABLE t1 (a VARCHAR(1) PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+connect stop_purge,localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+ALTER TABLE t1 ADD c INT;
+BEGIN;
+DELETE FROM t1;
+connect dml,localhost,root,,test;
+SET DEBUG_SYNC='row_mysql_handle_errors SIGNAL s1 WAIT_FOR s2';
+UPDATE t1 SET c=1;
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR s1';
+COMMIT;
+connection stop_purge;
+COMMIT;
+disconnect stop_purge;
+connection default;
+InnoDB	0 transactions not purged
+SET DEBUG_SYNC='now SIGNAL s2';
+connection dml;
+disconnect dml;
+connection default;
+SET DEBUG_SYNC=RESET;
+DROP TABLE t1;
 # End of 10.3 tests
 SET GLOBAL innodb_purge_rseg_truncate_frequency = @save_frequency;

mysql-test/suite/innodb/t/instant_alter_debug.test

@@ -364,6 +364,44 @@ DROP TABLE t1;
 
 SET GLOBAL innodb_limit_optimistic_insert_debug = @saved_limit;
 
+--echo #
+--echo # MDEV-24620 ASAN heap-buffer-overflow in btr_pcur_restore_position()
+--echo #
+
+CREATE TABLE t1 (a VARCHAR(1) PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+connect (stop_purge,localhost,root,,);
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+connection default;
+ALTER TABLE t1 ADD c INT;
+BEGIN;
+DELETE FROM t1;
+
+connect (dml,localhost,root,,test);
+SET DEBUG_SYNC='row_mysql_handle_errors SIGNAL s1 WAIT_FOR s2';
+send UPDATE t1 SET c=1;
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR s1';
+COMMIT;
+
+connection stop_purge;
+COMMIT;
+disconnect stop_purge;
+
+connection default;
+--source include/wait_all_purged.inc
+SET DEBUG_SYNC='now SIGNAL s2';
+
+connection dml;
+reap;
+disconnect dml;
+
+connection default;
+SET DEBUG_SYNC=RESET;
+DROP TABLE t1;
+
 --echo # End of 10.3 tests
 
 SET GLOBAL innodb_purge_rseg_truncate_frequency = @save_frequency;

storage/innobase/btr/btr0btr.cc

@@ -2,7 +2,7 @@
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2014, 2020, MariaDB Corporation.
+Copyright (c) 2014, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -890,7 +890,7 @@ btr_page_get_father_node_ptr_func(
 
 node_ptr = btr_cur_get_rec(cursor);
 
-offsets = rec_get_offsets(node_ptr, index, offsets, false,
+offsets = rec_get_offsets(node_ptr, index, offsets, 0,
  ULINT_UNDEFINED, &heap);
 
 if (btr_node_ptr_get_child_page_no(node_ptr, offsets) != page_no) {
@@ -907,10 +907,11 @@ btr_page_get_father_node_ptr_func(
 print_rec = page_rec_get_next(
 page_get_infimum_rec(page_align(user_rec)));
 offsets = rec_get_offsets(print_rec, index, offsets,
- page_rec_is_leaf(user_rec),
+ page_rec_is_leaf(user_rec)
+ ? index->n_core_fields : 0,
  ULINT_UNDEFINED, &heap);
 page_rec_print(print_rec, offsets);
-offsets = rec_get_offsets(node_ptr, index, offsets, false,
+offsets = rec_get_offsets(node_ptr, index, offsets, 0,
  ULINT_UNDEFINED, &heap);
 page_rec_print(node_ptr, offsets);
 
@@ -2214,7 +2215,9 @@ btr_page_get_split_rec(
 incl_data += insert_size;
 } else {
 offsets = rec_get_offsets(rec, cursor->index, offsets,
- page_is_leaf(page),
+ page_is_leaf(page)
+ ? cursor->index->n_core_fields
+ : 0,
  ULINT_UNDEFINED, &heap);
 incl_data += rec_offs_size(offsets);
 }
@@ -2323,7 +2326,9 @@ btr_page_insert_fits(
 space after rec is removed from page. */
 
 *offsets = rec_get_offsets(rec, cursor->index, *offsets,
- page_is_leaf(page),
+ page_is_leaf(page)
+ ? cursor->index->n_core_fields
+ : 0,
  ULINT_UNDEFINED, heap);
 
 total_data -= rec_offs_size(*offsets);
@@ -2610,7 +2615,8 @@ btr_page_tuple_smaller(
 first_rec = page_cur_get_rec(&pcur);
 
 *offsets = rec_get_offsets(
-first_rec, cursor->index, *offsets, page_is_leaf(block->frame),
+first_rec, cursor->index, *offsets,
+page_is_leaf(block->frame) ? cursor->index->n_core_fields : 0,
 n_uniq, heap);
 
 return(cmp_dtuple_rec(tuple, first_rec, *offsets) < 0);
@@ -2894,7 +2900,9 @@ btr_page_split_and_insert(
 first_rec = move_limit = split_rec;
 
 *offsets = rec_get_offsets(split_rec, cursor->index, *offsets,
- page_is_leaf(page), n_uniq, heap);
+ page_is_leaf(page)
+ ? cursor->index->n_core_fields : 0,
+ n_uniq, heap);
 
 insert_left = !tuple
 || cmp_dtuple_rec(tuple, split_rec, *offsets) < 0;
@@ -3665,7 +3673,7 @@ btr_compress(
 rec_offs*	offsets2 = NULL;
 
 /* For rtree, we need to update father's mbr. */
-if (dict_index_is_spatial(index)) {
+if (index->is_spatial()) {
 /* We only support merge pages with the same parent
 page */
 if (!rtr_check_same_block(
@@ -3683,7 +3691,8 @@ btr_compress(
 
 offsets2 = rec_get_offsets(
 btr_cur_get_rec(&cursor2), index, NULL,
-page_is_leaf(cursor2.page_cur.block->frame),
+page_is_leaf(cursor2.page_cur.block->frame)
+? index->n_fields : 0,
 ULINT_UNDEFINED, &heap);
 
 /* Check if parent entry needs to be updated */
@@ -3857,13 +3866,14 @@ btr_compress(
 #endif /* UNIV_DEBUG */
 
 /* For rtree, we need to update father's mbr. */
-if (dict_index_is_spatial(index)) {
+if (index->is_spatial()) {
 rec_offs* offsets2;
 ulint	rec_info;
 
 offsets2 = rec_get_offsets(
 btr_cur_get_rec(&cursor2), index, NULL,
-page_is_leaf(cursor2.page_cur.block->frame),
+page_is_leaf(cursor2.page_cur.block->frame)
+? index->n_fields : 0,
 ULINT_UNDEFINED, &heap);
 
 ut_ad(btr_node_ptr_get_child_page_no(
@@ -4341,7 +4351,7 @@ btr_print_recursive(
 node_ptr = page_cur_get_rec(&cursor);
 
 *offsets = rec_get_offsets(
-node_ptr, index, *offsets, false,
+node_ptr, index, *offsets, 0,
 ULINT_UNDEFINED, heap);
 btr_print_recursive(index,
  btr_node_ptr_get_child(node_ptr,
@@ -4490,7 +4500,9 @@ btr_index_rec_validate(
 
 page = page_align(rec);
 
-if (dict_index_is_ibuf(index)) {
+ut_ad(index->n_core_fields);
+
+if (index->is_ibuf()) {
 /* The insert buffer index tree can contain records from any
 other index: we cannot check the number of fields or
 their length */
@@ -4536,7 +4548,8 @@ btr_index_rec_validate(
 }
 }
 
-offsets = rec_get_offsets(rec, index, offsets, page_is_leaf(page),
+offsets = rec_get_offsets(rec, index, offsets, page_is_leaf(page)
+ ? index->n_core_fields : 0,
  ULINT_UNDEFINED, &heap);
 
 for (unsigned i = 0; i < index->n_fields; i++) {
@@ -4792,7 +4805,7 @@ btr_validate_level(
 page_cur_move_to_next(&cursor);
 
 node_ptr = page_cur_get_rec(&cursor);
-offsets = rec_get_offsets(node_ptr, index, offsets, false,
+offsets = rec_get_offsets(node_ptr, index, offsets, 0,
  ULINT_UNDEFINED, &heap);
 
 savepoint2 = mtr_set_savepoint(&mtr);
@@ -4916,10 +4929,12 @@ btr_validate_level(
 right_rec = page_rec_get_next(page_get_infimum_rec(
  right_page));
 offsets = rec_get_offsets(rec, index, offsets,
- page_is_leaf(page),
+ page_is_leaf(page)
+ ? index->n_core_fields : 0,
  ULINT_UNDEFINED, &heap);
 offsets2 = rec_get_offsets(right_rec, index, offsets2,
- page_is_leaf(right_page),
+ page_is_leaf(right_page)
+ ? index->n_core_fields : 0,
  ULINT_UNDEFINED, &heap);
 
 /* For spatial index, we cannot guarantee the key ordering

storage/innobase/btr/btr0bulk.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2014, 2019, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -193,7 +193,8 @@ PageBulk::insert(
 if (!page_rec_is_infimum_low(page_offset(m_cur_rec))) {
 rec_t*	old_rec = m_cur_rec;
 rec_offs* old_offsets = rec_get_offsets(
-old_rec, m_index, NULL,	is_leaf,
+old_rec, m_index, NULL,	is_leaf
+? m_index->n_core_fields : 0,
 ULINT_UNDEFINED, &m_heap);
 
 ut_ad(cmp_rec_rec(rec, old_rec, offsets, old_offsets, m_index)
@@ -447,6 +448,7 @@ PageBulk::getSplitRec()
 
 ut_ad(m_page_zip != NULL);
 ut_ad(m_rec_no >= 2);
+ut_ad(!m_index->is_instant());
 
 ut_ad(page_get_free_space_of_empty(m_is_comp) > m_free_space);
 total_used_size = page_get_free_space_of_empty(m_is_comp)
@@ -456,13 +458,13 @@ PageBulk::getSplitRec()
 n_recs = 0;
 offsets = NULL;
 rec = page_get_infimum_rec(m_page);
+const ulint n_core = page_is_leaf(m_page) ? m_index->n_core_fields : 0;
 
 do {
 rec = page_rec_get_next(rec);
 ut_ad(page_rec_is_user_rec(rec));
 
-offsets = rec_get_offsets(rec, m_index, offsets,
- page_is_leaf(m_page),
+offsets = rec_get_offsets(rec, m_index, offsets, n_core,
  ULINT_UNDEFINED, &m_heap);
 total_recs_size += rec_offs_size(offsets);
 n_recs++;
@@ -491,9 +493,11 @@ PageBulk::copyIn(
 ut_ad(m_rec_no == 0);
 ut_ad(page_rec_is_user_rec(rec));
 
+const ulint n_core = page_rec_is_leaf(rec)
+? m_index->n_core_fields : 0;
+
 do {
-offsets = rec_get_offsets(rec, m_index, offsets,
- page_rec_is_leaf(split_rec),
+offsets = rec_get_offsets(rec, m_index, offsets, n_core,
  ULINT_UNDEFINED, &m_heap);
 
 insert(rec, offsets);
@@ -534,17 +538,18 @@ PageBulk::copyOut(
 /* Set last record's next in page */
 rec_offs*	offsets = NULL;
 rec = page_rec_get_prev(split_rec);
-offsets = rec_get_offsets(rec, m_index, offsets,
- page_rec_is_leaf(split_rec),
+const ulint n_core = page_rec_is_leaf(split_rec)
+? m_index->n_core_fields : 0;
+
+offsets = rec_get_offsets(rec, m_index, offsets, n_core,
  ULINT_UNDEFINED, &m_heap);
 page_rec_set_next(rec, page_get_supremum_rec(m_page));
 
 /* Set related members */
 m_cur_rec = rec;
 m_heap_top = rec_get_end(rec, offsets);
 
-offsets = rec_get_offsets(last_rec, m_index, offsets,
- page_rec_is_leaf(split_rec),
+offsets = rec_get_offsets(last_rec, m_index, offsets, n_core,
  ULINT_UNDEFINED, &m_heap);
 
 m_free_space += ulint(rec_get_end(last_rec, offsets) - m_heap_top)
@@ -975,7 +980,8 @@ BtrBulk::insert(
 /* Convert tuple to rec. */
  rec = rec_convert_dtuple_to_rec(static_cast<byte*>(mem_heap_alloc(
 page_bulk->m_heap, rec_size)), m_index, tuple, n_ext);
- offsets = rec_get_offsets(rec, m_index, offsets, !level,
+ offsets = rec_get_offsets(rec, m_index, offsets, level
+ ? 0 : m_index->n_core_fields,
  ULINT_UNDEFINED, &page_bulk->m_heap);
 
 page_bulk->insert(rec, offsets);