MDEV-27272 Crash on EXPORT/IMPORT tablespace with column added in the middle

dict_index_t::reconstruct_fields(): add input validation by replacing some
assertions

handle_instant_metadata(): fix nullptr dereference

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

@@ -118,3 +118,12 @@ FLUSH TABLE t1 FOR EXPORT;
 UNLOCK TABLES;
 ALTER TABLE t2 IMPORT TABLESPACE;
 DROP TABLE t2, t1;
+CREATE TABLE t1 ( id INT NOT NULL, i1 INT, i2 INT, PRIMARY KEY (id)) engine=innodb;
+CREATE TABLE t2 ( id INT NOT NULL, i1 INT, i2 INT, PRIMARY KEY (id)) engine=innodb;
+ALTER TABLE test.t1 add COLUMN i3 INT AFTER i1;
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR HY000: Index for table 't2' is corrupt; try to repair it
+DROP TABLE t1, t2;

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

@@ -2,6 +2,11 @@
 --source include/have_sequence.inc
 --source include/innodb_checksum_algorithm.inc
 
+--disable_query_log
+call mtr.add_suppression("Table `test`.`t2` contains unrecognizable instant ALTER metadata");
+call mtr.add_suppression("Index for table 't2' is corrupt; try to repair it");
+--enable_query_log
+
 set default_storage_engine=innodb;
 
 --echo #
@@ -180,3 +185,21 @@ UNLOCK TABLES;
 ALTER TABLE t2 IMPORT TABLESPACE;
 
 DROP TABLE t2, t1;
+
+
+CREATE TABLE t1 ( id INT NOT NULL, i1 INT, i2 INT, PRIMARY KEY (id)) engine=innodb;
+CREATE TABLE t2 ( id INT NOT NULL, i1 INT, i2 INT, PRIMARY KEY (id)) engine=innodb;
+
+ALTER TABLE test.t1 add COLUMN i3 INT AFTER i1;
+
+ALTER TABLE t2 DISCARD TABLESPACE;
+FLUSH TABLES t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+UNLOCK TABLES;
+--error ER_NOT_KEYFILE
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+DROP TABLE t1, t2;

storage/innobase/dict/dict0mem.cc

@@ -1211,8 +1211,9 @@ bool dict_foreign_t::affects_fulltext() const
   return false;
 }
 
-/** Reconstruct the clustered index fields. */
-inline void dict_index_t::reconstruct_fields()
+/** Reconstruct the clustered index fields.
+@return whether metadata is incorrect */
+inline bool dict_index_t::reconstruct_fields()
 {
 	DBUG_ASSERT(is_primary());
 
@@ -1243,10 +1244,14 @@ inline void dict_index_t::reconstruct_fields()
 				fields + n_first, fields + n_fields,
 				[c](const dict_field_t& o)
 				{ return o.col->ind == c.ind(); });
+
+			if (old >= fields + n_fields
+			    || old->prefix_len
+			    || old->col != &table->cols[c.ind()]) {
+				return true;
+			}
+
 			ut_ad(old >= &fields[n_first]);
-			ut_ad(old < &fields[n_fields]);
-			DBUG_ASSERT(!old->prefix_len);
-			DBUG_ASSERT(old->col == &table->cols[c.ind()]);
 			f = *old;
 		}
 
@@ -1259,6 +1264,8 @@ inline void dict_index_t::reconstruct_fields()
 
 	fields = tfields;
 	n_core_null_bytes = UT_BITS_IN_BYTES(n_core_null);
+
+	return false;
 }
 
 /** Reconstruct dropped or reordered columns.
@@ -1323,8 +1330,7 @@ bool dict_table_t::deserialise_columns(const byte* metadata, ulint len)
 	}
 	DBUG_ASSERT(col == &dropped_cols[n_dropped_cols]);
 
-	UT_LIST_GET_FIRST(indexes)->reconstruct_fields();
-	return false;
+	return UT_LIST_GET_FIRST(indexes)->reconstruct_fields();
 }
 
 /** Check if record in clustered index is historical row.

storage/innobase/include/dict0mem.h

@@ -1321,8 +1321,9 @@ struct dict_index_t {
   ulint get_new_n_vcol() const
   { return new_vcol_info ? new_vcol_info->n_v_col : 0; }
 
-  /** Reconstruct the clustered index fields. */
-  inline void reconstruct_fields();
+  /** Reconstruct the clustered index fields.
+  @return whether metadata is incorrect */
+  inline bool reconstruct_fields();
 
   /** Check if the index contains a column or a prefix of that column.
   @param[in]	n		column number

storage/innobase/row/row0import.cc

@@ -3288,7 +3288,10 @@ static dberr_t handle_instant_metadata(dict_table_t *table,
     }
 
     mem_heap_t *heap= NULL;
-    SCOPE_EXIT([&heap]() { mem_heap_free(heap); });
+    SCOPE_EXIT([&heap]() {
+      if (heap)
+        mem_heap_free(heap);
+    });
 
     while (btr_page_get_level(page.get()) != 0)
     {