MDEV-35237 Bulk insert fails to apply buffered operation during CREATE..SELECT statement

Problem:
=======
- InnoDB fails to write the buffered insert operation during
create..select operation. This happens when bulk_insert
in transaction is reset to false while unlocking a source table.

Fix:
===
- InnoDB should apply the previous buffered changes to
all tables if we encounter any statement other than
pure INSERT or INSERT..SELECT statement in
ha_innobase::external_lock() and start_stmt().

- Remove the function bulk_insert_apply_for_table()

start_stmt(), external_lock(): Assert that trx->duplicates
should be enabled during bulk insert operation

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

@@ -50,4 +50,20 @@ ALTER TABLE t2 ALGORITHM=COPY, FORCE;
 affected rows: 1
 info: Records: 1  Duplicates: 0  Warnings: 0
 DROP TABLE t2, t1;
+#
+#  MDEV-35237 Bulk insert fails to apply buffered
+#              operation during copy alter
+#
+CREATE TABLE t1 (f1 int NOT NULL, f2 tinyint(1) NOT NULL,
+f3 varchar(80) NOT NULL, PRIMARY KEY(f1),
+KEY(f2), KEY(f3))ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1,1,''),(2,0,''), (4,1,'e');
+CREATE TABLE t2 (f1 int NOT NULL, f2 int NOT NULL,KEY(f1))ENGINE=InnoDB;
+INSERT INTO t2 VALUES (1,0),(1,0);
+CREATE TABLE t engine=innodb
+SELECT t2.f2 FROM t2 JOIN t1 ON t1.f1 = t2.f1 AND t1.f3 = '' AND t1.f2=1 ;
+SELECT COUNT(*) FROM t;
+COUNT(*)
+2
+DROP TABLE t1, t2, t;
 SET GLOBAL innodb_stats_persistent=@default_stats_persistent;

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

@@ -68,4 +68,19 @@ ALTER TABLE t1 ALGORITHM=COPY, FORCE;
 ALTER TABLE t2 ALGORITHM=COPY, FORCE;
 --disable_info
 DROP TABLE t2, t1;
+
+--echo #
+--echo #  MDEV-35237 Bulk insert fails to apply buffered
+--echo #              operation during copy alter
+--echo #
+CREATE TABLE t1 (f1 int NOT NULL, f2 tinyint(1) NOT NULL,
+                 f3 varchar(80) NOT NULL, PRIMARY KEY(f1),
+                 KEY(f2), KEY(f3))ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1,1,''),(2,0,''), (4,1,'e');
+CREATE TABLE t2 (f1 int NOT NULL, f2 int NOT NULL,KEY(f1))ENGINE=InnoDB;
+INSERT INTO t2 VALUES (1,0),(1,0);
+CREATE TABLE t engine=innodb
+SELECT t2.f2 FROM t2 JOIN t1 ON t1.f1 = t2.f1 AND t1.f3 = '' AND t1.f2=1 ;
+SELECT COUNT(*) FROM t;
+DROP TABLE t1, t2, t;
 SET GLOBAL innodb_stats_persistent=@default_stats_persistent;

storage/innobase/handler/ha_innodb.cc

@@ -15902,20 +15902,17 @@ ha_innobase::start_stmt(
 		if (trx->is_bulk_insert()) {
 			/* Allow a subsequent INSERT into an empty table
 			if !unique_checks && !foreign_key_checks. */
+			ut_ad(!trx->duplicates);
 			break;
 		}
 		/* fall through */
 	default:
-		trx->bulk_insert_apply_for_table(m_prebuilt->table);
+		trx->bulk_insert_apply();
+		trx->end_bulk_insert();
 		if (!trx->bulk_insert) {
 			break;
 		}
 
-		/* Trigger could've initiated another stmt.
-		So apply all bulk operation and mark as
-		end bulk insert for all tables */
-		trx->bulk_insert_apply();
-		trx->end_bulk_insert();
 		trx->bulk_insert = false;
 		trx->last_sql_stat_start.least_undo_no = trx->undo_no;
 	}
@@ -16096,11 +16093,13 @@ ha_innobase::external_lock(
 		if (trx->is_bulk_insert()) {
 			/* Allow a subsequent INSERT into an empty table
 			if !unique_checks && !foreign_key_checks. */
+			ut_ad(!trx->duplicates);
 			break;
 		}
 		/* fall through */
 	default:
-		trx->bulk_insert_apply_for_table(m_prebuilt->table);
+		trx->bulk_insert_apply();
+		trx->end_bulk_insert();
 		if (!trx->bulk_insert) {
 			break;
 		}

storage/innobase/include/trx0trx.h

@@ -1172,10 +1172,6 @@ public:
     return UNIV_UNLIKELY(bulk_insert) ? bulk_insert_apply_low(): DB_SUCCESS;
   }
 
-  /** Do the bulk insert for the buffered insert operation of a table.
-  @param table bulk insert operation
-  @return DB_SUCCESS or error code. */
-  dberr_t bulk_insert_apply_for_table(dict_table_t *table);
 private:
   /** Apply the buffered bulk inserts. */
   dberr_t bulk_insert_apply_low();

storage/innobase/row/row0merge.cc

@@ -5345,21 +5345,6 @@ void trx_t::bulk_rollback_low()
   rollback(&bulk_save);
 }
 
-dberr_t trx_t::bulk_insert_apply_for_table(dict_table_t *table)
-{
-  auto it= mod_tables.find(table);
-  if (it != mod_tables.end())
-  {
-    if (dberr_t err= it->second.write_bulk(table, this))
-    {
-      bulk_rollback_low();
-      return err;
-    }
-    it->second.end_bulk_insert();
-  }
-  return DB_SUCCESS;
-}
-
 dberr_t trx_t::bulk_insert_apply_low()
 {
   ut_ad(bulk_insert);