MDEV-22970 Possible corruption of page_compressed tables, or

           when scrubbing is enabled

buf_read_recv_pages(): Ignore the page to read if it is already
present in the freed ranges.

store_freed_or_init_rec(): Store the ranges only if scrubbing
is enabled or page compressed tablespace.

recv_init_crash_recovery_space(): Add the freed range only when
scrubbing or page compressed tablespace.

range_set::contains(): Search the value is present in ranges.

range_set::remove_if_exists(): Remove the value if exist in ranges.

mtr_t::init(): Handles the scenario that mini-transaction may allocate
a page that had just been freed.

recv_sys_t::parse(): Note down the FREE and INIT redo log irrespective
of STORE value.

Removed innodb_tablespaces_scrubbing from test case

mysql-test/main/information_schema_all_engines-master.opt

@@ -15,4 +15,3 @@
 --loose-innodb-sys-tablestats
 --loose-innodb-mutexes
 --loose-innodb-tablespaces-encryption
---loose-innodb-tablespaces-scrubbing

mysql-test/suite/innodb/disabled.def

@@ -12,4 +12,3 @@
 
 create-index-debug : MDEV-13680 InnoDB may crash when btr_page_alloc() fails
 innodb_force_recovery_rollback : MDEV-22889 InnoDB occasionally breaks ACID
-innodb_scrub : MDEV-8139/MDEV-22970 Fix scrubbing

mysql-test/suite/innodb/t/innodb_skip_innodb_is_tables.opt

@@ -59,9 +59,7 @@
 --loose-innodb_sys_datafiles
 --loose-innodb_changed_pages
 --loose-innodb_tablespaces_encryption
---loose-innodb_tablespaces_scrubbing
 --loose-innodb_mutexes
 --loose-innodb_sys_semaphore_waits
---loose-innodb_tablespaces_scrubbing
 --loose-innodb_mutexes
 --loose-innodb_sys_semaphore_waits

storage/innobase/buf/buf0buf.cc

@@ -2656,14 +2656,9 @@ void buf_page_free(const page_id_t page_id,
   buf_block_t *block= reinterpret_cast<buf_block_t*>
     (buf_pool.page_hash_get_low(page_id, fold));
 
-#if 0 /* FIXME: MDEV-22970 Potential corruption */
-  /* TODO: Find out how and when a freed page can be marked
-  allocated in the same mini-transaction. At least it seems to
-  happen during a pessimistic insert operation. */
   /* TODO: try to all this part of mtr_t::free() */
   if (srv_immediate_scrub_data_uncompressed || mtr->is_page_compressed())
     mtr->add_freed_offset(page_id);
-#endif
 
   if (!block || block->page.state() != BUF_BLOCK_FILE_PAGE)
   {

storage/innobase/buf/buf0rea.cc

@@ -746,6 +746,13 @@ buf_read_recv_pages(
 	const ulint zip_size = space->zip_size();
 
 	for (ulint i = 0; i < n_stored; i++) {
+
+		/* Ignore if the page already present in freed ranges. */
+		if (space->freed_ranges.contains(
+			static_cast<uint32_t>(page_nos[i]))) {
+			continue;
+		}
+
 		const page_id_t	cur_page_id(space_id, page_nos[i]);
 
 		ulint limit = 0;

storage/innobase/handler/i_s.h

@@ -63,7 +63,6 @@ extern struct st_maria_plugin	i_s_innodb_sys_datafiles;
 extern struct st_maria_plugin	i_s_innodb_mutexes;
 extern struct st_maria_plugin	i_s_innodb_sys_virtual;
 extern struct st_maria_plugin	i_s_innodb_tablespaces_encryption;
-extern struct st_maria_plugin	i_s_innodb_tablespaces_scrubbing;
 extern struct st_maria_plugin	i_s_innodb_sys_semaphore_waits;
 
 /** The latest successfully looked up innodb_fts_aux_table */

storage/innobase/include/fil0fil.h

@@ -132,6 +132,20 @@ class range_set
 {
 private:
   range_set_t ranges;
+
+  range_set_t::iterator find(uint32_t value) const
+  {
+    auto r_offset= ranges.lower_bound({value, value});
+    const auto r_end= ranges.end();
+    if (r_offset != r_end);
+    else if (empty())
+      return r_end;
+    else
+      r_offset= std::prev(r_end);
+    if (r_offset->first <= value && r_offset->last >= value)
+      return r_offset;
+    return r_end;
+  }
 public:
   /** Merge the current range with previous range.
   @param[in] range      range to be merged
@@ -194,7 +208,7 @@ class range_set
   @param[in]	value	Value to be removed. */
   void remove_value(uint32_t value)
   {
-    if (ranges.empty())
+    if (empty())
       return;
     range_t new_range {value, value};
     range_set_t::iterator range= ranges.lower_bound(new_range);
@@ -273,6 +287,22 @@ class range_set
     add_range(new_range);
   }
 
+  bool remove_if_exists(uint32_t value)
+  {
+    auto r_offset= find(value);
+    if (r_offset != ranges.end())
+    {
+      remove_within_range(r_offset, value);
+      return true;
+    }
+    return false;
+  }
+
+  bool contains(uint32_t value) const
+  {
+    return find(value) != ranges.end();
+  }
+
   ulint size() { return ranges.size(); }
   void clear() { ranges.clear(); }
   bool empty() const { return ranges.empty(); }

storage/innobase/include/mtr0log.h

@@ -511,7 +511,17 @@ inline void mtr_t::memcpy(const buf_block_t &b, void *dest, const void *str,
 @param[in,out]        b       buffer page */
 inline void mtr_t::init(buf_block_t *b)
 {
-  ut_ad(!m_freed_pages);
+  if (UNIV_LIKELY_NULL(m_freed_pages))
+  {
+    ut_ad(m_user_space->id == b->page.id().space());
+    if (m_freed_pages->remove_if_exists(b->page.id().page_no()) &&
+        m_freed_pages->empty())
+    {
+      delete m_freed_pages;
+      m_freed_pages= nullptr;
+    }
+  }
+
   b->page.status= buf_page_t::INIT_ON_FLUSH;
 
   if (m_log_mode != MTR_LOG_ALL)

storage/innobase/log/log0recv.cc

@@ -1779,7 +1779,6 @@ inline void recv_sys_t::add(const page_id_t page_id,
                   log_phys_t(start_lsn, lsn, l, len));
 }
 
-#if 0 /* FIXME: MDEV-22970 Potential corruption */
 /** Store/remove the freed pages in fil_name_t of recv_spaces.
 @param[in]	page_id		freed or init page_id
 @param[in]	freed		TRUE if page is freed */
@@ -1789,6 +1788,8 @@ static void store_freed_or_init_rec(page_id_t page_id, bool freed)
   uint32_t page_no= page_id.page_no();
   if (is_predefined_tablespace(space_id))
   {
+    if (!srv_immediate_scrub_data_uncompressed)
+      return;
     fil_space_t *space;
     if (space_id == TRX_SYS_SPACE)
       space= fil_system.sys_space;
@@ -1808,7 +1809,6 @@ static void store_freed_or_init_rec(page_id_t page_id, bool freed)
       i->second.remove_freed_page(page_no);
   }
 }
-#endif
 
 /** Parse and register one mini-transaction in log_t::FORMAT_10_5.
 @param checkpoint_lsn  the log sequence number of the latest checkpoint
@@ -2008,9 +2008,7 @@ bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t *store, bool apply)
       case INIT_PAGE:
         last_offset= FIL_PAGE_TYPE;
       free_or_init_page:
-#if 0 /* FIXME: MDEV-22970 Potential corruption */
         store_freed_or_init_rec(id, (b & 0x70) == FREE_PAGE);
-#endif
         if (UNIV_UNLIKELY(rlen != 0))
           goto record_corrupted;
         break;
@@ -2135,12 +2133,12 @@ bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t *store, bool apply)
       case STORE_NO:
         if (!is_init)
           continue;
+        mlog_init.add(id, start_lsn);
         map::iterator i= pages.find(id);
         if (i == pages.end())
           continue;
         i->second.log.clear();
         pages.erase(i);
-        mlog_init.add(id, start_lsn);
       }
     }
 #if 1 /* MDEV-14425 FIXME: this must be in the checkpoint file only! */
@@ -3291,9 +3289,13 @@ recv_init_crash_recovery_spaces(bool rescan, bool& missing_tablespace)
 
 			/* Add the freed page ranges in the respective
 			tablespace */
-			if (!rs.second.freed_ranges.empty())
-			  rs.second.space->add_free_ranges(
+			if (!rs.second.freed_ranges.empty()
+			    && (srv_immediate_scrub_data_uncompressed
+				|| rs.second.space->is_compressed())) {
+
+				rs.second.space->add_free_ranges(
 					std::move(rs.second.freed_ranges));
+			}
 		} else if (rs.second.name == "") {
 			ib::error() << "Missing FILE_CREATE, FILE_DELETE"
 				" or FILE_MODIFY before FILE_CHECKPOINT"