MDEV-35830 Fix innodb_undo_log_truncate in backup

recv_sys_t::parse(): Correctly handle the storing==BACKUP case,
and simplify some logic around storing==YES as well.

The added test mariabackup.undo_truncate is based on an idea of
Thirunarayanan Balathandayuthapani. It nondeterministically (not on
every run) covers this logic, including the function backup_undo_trunc(),
for both innodb_encrypt_log=ON and innodb_encrypt_log=OFF.

Reviewed by: Debarun Banerjee

mysql-test/suite/mariabackup/undo_truncate.combinations

@@ -0,0 +1,4 @@
+[clear]
+--innodb-encrypt-log=OFF
+[crypt]
+--innodb-encrypt-log=ON

mysql-test/suite/mariabackup/undo_truncate.opt

@@ -0,0 +1,6 @@
+--innodb-undo-tablespaces=2
+--plugin-load-add=$FILE_KEY_MANAGEMENT_SO
+--loose-file-key-management
+--loose-file-key-management-filekey=FILE:$MTR_SUITE_DIR/filekeys-data.key
+--loose-file-key-management-filename=$MTR_SUITE_DIR/filekeys-data.enc
+--loose-file-key-management-encryption-algorithm=aes_cbc

mysql-test/suite/mariabackup/undo_truncate.result

@@ -0,0 +1,39 @@
+SET GLOBAL innodb_undo_log_truncate = 0;
+create table t1 (keyc int primary key default 0, c char(6)) engine=innodb;
+create table t2 (keyc int primary key default 0, c char(6)) engine=innodb;
+CREATE PROCEDURE p(t VARCHAR(64))
+BEGIN
+DECLARE i TEXT DEFAULT 'insert into t1 select seq,repeat(chr(48),6)
+    from seq_1_to_20000';
+DECLARE u1 TEXT DEFAULT 'update t1 set c=repeat(chr(32),6)';
+DECLARE u2 TEXT DEFAULT 'update t1 set c=repeat(chr(64),6)';
+EXECUTE IMMEDIATE REPLACE(i,'t1', t);
+EXECUTE IMMEDIATE REPLACE(u1,'t1', t);
+EXECUTE IMMEDIATE REPLACE(u2,'t1', t);
+END;
+$$
+connect  con1,localhost,root,,;
+begin;
+call p('t1');
+connection default;
+call p('t2');
+connection con1;
+commit;
+disconnect con1;
+connection default;
+DROP PROCEDURE p;
+SET GLOBAL innodb_undo_log_truncate = 1;
+SET GLOBAL innodb_max_undo_log_size=DEFAULT;
+SET GLOBAL innodb_max_purge_lag_wait=0;
+# Prepare full backup
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart
+select count(*) from t1;
+count(*)
+20000
+select count(*) from t2;
+count(*)
+20000
+DROP TABLE t1,t2;

mysql-test/suite/mariabackup/undo_truncate.test

@@ -0,0 +1,59 @@
+--source include/have_innodb.inc
+--source include/not_embedded.inc
+--source include/have_sequence.inc
+--source include/have_file_key_management.inc
+
+SET GLOBAL innodb_undo_log_truncate = 0;
+
+#
+# Perform DML action using multiple clients and multiple undo tablespace.
+#
+
+create table t1 (keyc int primary key default 0, c char(6)) engine=innodb;
+create table t2 (keyc int primary key default 0, c char(6)) engine=innodb;
+
+DELIMITER $$;
+CREATE PROCEDURE p(t VARCHAR(64))
+BEGIN
+  DECLARE i TEXT DEFAULT 'insert into t1 select seq,repeat(chr(48),6)
+    from seq_1_to_20000';
+  DECLARE u1 TEXT DEFAULT 'update t1 set c=repeat(chr(32),6)';
+  DECLARE u2 TEXT DEFAULT 'update t1 set c=repeat(chr(64),6)';
+  EXECUTE IMMEDIATE REPLACE(i,'t1', t);
+  EXECUTE IMMEDIATE REPLACE(u1,'t1', t);
+  EXECUTE IMMEDIATE REPLACE(u2,'t1', t);
+END;
+$$
+DELIMITER ;$$
+
+connect (con1,localhost,root,,);
+begin;
+send call p('t1');
+
+connection default;
+call p('t2');
+
+connection con1;
+reap;
+commit;
+disconnect con1;
+connection default;
+DROP PROCEDURE p;
+
+SET GLOBAL innodb_undo_log_truncate = 1;
+SET GLOBAL innodb_max_undo_log_size=DEFAULT;
+SET GLOBAL innodb_max_purge_lag_wait=0;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --parallel=10 --target-dir=$targetdir --throttle=1000;
+--echo # Prepare full backup
+exec $XTRABACKUP --prepare --target-dir=$targetdir;
+--enable_result_log
+
+source include/restart_and_restore.inc;
+select count(*) from t1;
+select count(*) from t2;
+# Cleanup
+rmdir $targetdir;
+DROP TABLE t1,t2;

storage/innobase/log/log0recv.cc

@@ -2698,43 +2698,57 @@ restart:
       mach_write_to_4(iv + 8, space_id);
       mach_write_to_4(iv + 12, page_no);
     }
-    got_page_op= !(b & 0x80);
-    if (!got_page_op);
-    else if (storing == BACKUP && srv_operation == SRV_OPERATION_BACKUP)
-    {
-      if (page_no == 0 && (b & 0xf0) == INIT_PAGE && first_page_init)
-        first_page_init(space_id);
-      continue;
-    }
-    else if (storing == YES && file_checkpoint &&
-             space_id != TRX_SYS_SPACE && !srv_is_undo_tablespace(space_id))
-    {
-      recv_spaces_t::iterator i= recv_spaces.lower_bound(space_id);
-      if (i != recv_spaces.end() && i->first == space_id);
-      else if (lsn < file_checkpoint)
-        /* We have not seen all records between the checkpoint and
-        FILE_CHECKPOINT. There should be a FILE_DELETE for this
-        tablespace later. */
-        recv_spaces.emplace_hint(i, space_id, file_name_t("", false));
-      else
-      {
-        const page_id_t id(space_id, page_no);
-        if (!srv_force_recovery)
-        {
-          ib::error() << "Missing FILE_DELETE or FILE_MODIFY for " << id
-                      << " at " << lsn
-                      << "; set innodb_force_recovery=1 to ignore the record.";
-          goto corrupted;
-        }
-        ib::warn() << "Ignoring record for " << id << " at " << lsn;
-        continue;
-      }
-    }
     DBUG_PRINT("ib_log",
                ("scan " LSN_PF ": rec %x len %zu page %u:%u",
                 lsn, b, l - recs + rlen, space_id, page_no));
+    got_page_op= !(b & 0x80);
     if (got_page_op)
     {
+      if (storing == BACKUP)
+      {
+        if (page_no == 0 && (b & 0xf0) == INIT_PAGE && first_page_init)
+          first_page_init(space_id);
+        else if (rlen == 1 && undo_space_trunc)
+        {
+          mach_write_to_4(iv + 8, space_id);
+          mach_write_to_4(iv + 12, page_no);
+          byte eb[1/*type,length*/ + 5/*space_id*/ + 5/*page_no*/ + 1/*rlen*/];
+          if (*l.copy_if_needed(iv, eb, recs, 1) == TRIM_PAGES)
+            undo_space_trunc(space_id);
+        }
+        continue;
+      }
+      if (storing == YES && UNIV_LIKELY(space_id != TRX_SYS_SPACE) &&
+          !srv_is_undo_tablespace(space_id))
+      {
+        ut_ad(file_checkpoint != 0);
+        recv_spaces_t::iterator i= recv_spaces.lower_bound(space_id);
+        if (i != recv_spaces.end() && i->first == space_id);
+        else if (lsn < file_checkpoint)
+          /* We have not seen all records between the checkpoint and
+          FILE_CHECKPOINT. There should be a FILE_DELETE for this
+          tablespace later. */
+          recv_spaces.emplace_hint(i, space_id, file_name_t("", false));
+        else
+        {
+          if (!srv_force_recovery)
+          {
+            sql_print_error("InnoDB: Missing FILE_DELETE or FILE_MODIFY for "
+                            "[page id: space=" UINT32PF
+                            ", page number=" UINT32PF "]"
+                            " at " LSN_PF
+                            "; set innodb_force_recovery=1 to"
+                            " ignore the record.",
+                            space_id, page_no, lsn);
+            goto corrupted;
+          }
+          sql_print_warning("InnoDB: Ignoring record for "
+                            "[page id: space=" UINT32PF
+                            ", page number=" UINT32PF "] at " LSN_PF,
+                            space_id, page_no, lsn);
+          continue;
+        }
+      }
     same_page:
       if (!rlen);
       else if (UNIV_UNLIKELY(l - recs + rlen > srv_page_size))
@@ -2748,13 +2762,11 @@ restart:
       case FREE_PAGE:
         ut_ad(freed.emplace(id).second);
         /* the next record must not be same_page */
-        if (storing != BACKUP) last_offset= 1;
+        last_offset= 1;
         goto free_or_init_page;
       case INIT_PAGE:
-        if (storing != BACKUP) last_offset= FIL_PAGE_TYPE;
+        last_offset= FIL_PAGE_TYPE;
       free_or_init_page:
-        if (storing == BACKUP)
-          continue;
         if (UNIV_UNLIKELY(rlen != 0))
           goto record_corrupted;
         store_freed_or_init_rec(id, (b & 0x70) == FREE_PAGE);
@@ -2799,17 +2811,6 @@ restart:
           continue;
         if (UNIV_UNLIKELY(!rlen))
           goto record_corrupted;
-        if (storing == BACKUP)
-        {
-          if (rlen == 1 && undo_space_trunc)
-          {
-            cl= l.copy_if_needed(iv, decrypt_buf, recs, rlen);
-            if (*cl == TRIM_PAGES)
-              undo_space_trunc(space_id);
-          }
-          continue;
-        }
-
         cl= l.copy_if_needed(iv, decrypt_buf, recs, rlen);
         if (rlen == 1 && *cl == TRIM_PAGES)
         {
@@ -2823,10 +2824,9 @@ restart:
           trim({space_id, 0}, start_lsn);
           truncated_undo_spaces[space_id - srv_undo_space_id_start]=
             { start_lsn, page_no };
-          if (storing != BACKUP)
-            /* the next record must not be same_page */
-            last_offset= 1;
-          else if (undo_space_trunc)
+          /* the next record must not be same_page */
+          last_offset= 1;
+          if (undo_space_trunc)
             undo_space_trunc(space_id);
           continue;
         }
@@ -2850,8 +2850,6 @@ restart:
       case WRITE:
       case MEMMOVE:
       case MEMSET:
-        if (storing == BACKUP)
-          continue;
         if (storing == NO && UNIV_LIKELY(page_no != 0))
           /* fil_space_set_recv_size_and_flags() is mandatory for storing==NO.
           It is only applicable to page_no == 0. Other than that, we can just