MDEV-19556: InnoDB native sampling

sql/handler.cc

@@ -3379,6 +3379,40 @@ int handler::ha_close(void)
   DBUG_RETURN(close());
 }
 
+int handler::ha_sample_next(uchar *buf)
+{
+  int result;
+  DBUG_ENTER("handler::ha_sample_next");
+  DBUG_ASSERT(table_share->tmp_table != NO_TMP_TABLE ||
+              m_lock_type != F_UNLCK);
+  DBUG_ASSERT(inited == SAMPLING);
+
+  do
+  {
+    TABLE_IO_WAIT(tracker, PSI_TABLE_FETCH_ROW, MAX_KEY, result,
+                  { result= sample_next(buf); })
+    if (result != HA_ERR_RECORD_DELETED && result != HA_ERR_KEY_NOT_FOUND)
+      break;
+    //status_var_increment(table->in_use->status_var.ha_read_rnd_deleted_count);
+  } while (!table->in_use->check_killed(1));
+
+  if (result == HA_ERR_RECORD_DELETED)
+    result= HA_ERR_ABORTED_BY_USER;
+  else
+  {
+    if (!result)
+    {
+      update_rows_read();
+      if (table->vfield && buf == table->record[0])
+        table->update_virtual_fields(this, VCOL_UPDATE_FOR_READ);
+    }
+    //increment_statistics(&SSV::ha_read_rnd_next_count);
+  }
+
+  table->status=result ? STATUS_NOT_FOUND: 0;
+  DBUG_RETURN(result);
+}
+
 
 int handler::ha_rnd_next(uchar *buf)
 {

sql/handler.h

@@ -366,7 +366,10 @@ enum chf_create_flags {
 /* Implements SELECT ... FOR UPDATE SKIP LOCKED */
 #define HA_CAN_SKIP_LOCKED  (1ULL << 61)
 
-#define HA_LAST_TABLE_FLAG HA_CAN_SKIP_LOCKED
+/* Implements native sampling (MDEV-19556) */
+#define HA_NATIVE_SAMPLING (1ULL << 62)
+
+#define HA_LAST_TABLE_FLAG HA_NATIVE_SAMPLING
 
 
 /* bits in index_flags(index_number) for what you can do with index */
@@ -3245,7 +3248,7 @@ class handler :public Sql_alloc
   /** Length of ref (1-8 or the clustered key length) */
   uint ref_length;
   FT_INFO *ft_handler;
-  enum init_stat { NONE=0, INDEX, RND };
+  enum init_stat { NONE=0, INDEX, RND, SAMPLING };
   init_stat inited, pre_inited;
 
   const COND *pushed_cond;
@@ -3506,6 +3509,25 @@ class handler :public Sql_alloc
   }
   int ha_rnd_init_with_error(bool scan) __attribute__ ((warn_unused_result));
   int ha_reset();
+  int ha_sample_init()
+  {
+    int result;
+    DBUG_ENTER("ha_sample_init");
+    DBUG_ASSERT(inited == NONE);
+    result= sample_init();
+    inited= result ? NONE: SAMPLING;
+    end_range= NULL;
+    DBUG_RETURN(result);
+  }
+  int ha_sample_next(uchar* buf);
+  int ha_sample_end()
+  {
+    DBUG_ENTER("ha_sample_end");
+    DBUG_ASSERT(inited==SAMPLING);
+    inited= NONE;
+    end_range= NULL;
+    DBUG_RETURN(sample_end());
+  }
   /* this is necessary in many places, e.g. in HANDLER command */
   int ha_index_or_rnd_end()
   {
@@ -4030,6 +4052,11 @@ class handler :public Sql_alloc
   virtual int ft_read(uchar *buf) { return HA_ERR_WRONG_COMMAND; }
   virtual int rnd_next(uchar *buf)=0;
   virtual int rnd_pos(uchar * buf, uchar *pos)=0;
+
+  virtual int sample_init() {return 0;}
+  virtual int sample_next(uchar *buf) { return 0;}
+  virtual int sample_end() {return 0;}
+
   /**
     This function only works for handlers having
     HA_PRIMARY_KEY_REQUIRED_FOR_POSITION set.

sql/sql_statistics.cc

@@ -2650,29 +2650,28 @@ int collect_statistics_for_table(THD *thd, TABLE *table)
 
   for (field_ptr= table->field; *field_ptr; field_ptr++)
   {
-    table_field= *field_ptr;   
+    table_field= *field_ptr;
     if (!table_field->collected_stats)
-      continue; 
+      continue;
     table_field->collected_stats->init(thd, table_field);
   }
 
   restore_record(table, s->default_values);
 
-  /* Perform a full table scan to collect statistics on 'table's columns */
-  if (!(rc= file->ha_rnd_init(TRUE)))
+  if(file->ha_table_flags() & HA_NATIVE_SAMPLING)
   {
-    DEBUG_SYNC(table->in_use, "statistics_collection_start");
-
-    while ((rc= file->ha_rnd_next(table->record[0])) != HA_ERR_END_OF_FILE)
-    {
-      if (thd->killed)
-        break;
-
-      if (rc)
-        break;
+    if(!(rc= file->ha_sample_init())) {
+      DEBUG_SYNC(table->in_use, "statistics_collection_start");
+      rows = file->records() * (thd->variables.sample_percentage / 100);
 
-      if (thd_rnd(thd) <= sample_fraction)
+      for (ulonglong i= 0; i < rows; ++i)
       {
+        rc = file->ha_sample_next(table->record[0]);
+        if (thd->killed)
+          break;
+
+        if (rc)
+          break;
         for (field_ptr= table->field; *field_ptr; field_ptr++)
         {
           table_field= *field_ptr;
@@ -2683,12 +2682,47 @@ int collect_statistics_for_table(THD *thd, TABLE *table)
         }
         if (rc)
           break;
-        rows++;
       }
+      file->ha_sample_end();
     }
-    file->ha_rnd_end();
+    rc= (rc == 0 && !thd->killed) ? 0 : 1;
   }
-  rc= (rc == HA_ERR_END_OF_FILE && !thd->killed) ? 0 : 1;
+  else
+  {
+
+    /* Perform a full table scan to collect statistics on 'table's columns */
+    if (!(rc= file->ha_rnd_init(TRUE)))
+    {
+      DEBUG_SYNC(table->in_use, "statistics_collection_start");
+
+      while ((rc= file->ha_rnd_next(table->record[0])) != HA_ERR_END_OF_FILE)
+      {
+        if (thd->killed)
+          break;
+
+        if (rc)
+          break;
+
+        if (thd_rnd(thd) <= sample_fraction)
+        {
+          for (field_ptr= table->field; *field_ptr; field_ptr++)
+          {
+            table_field= *field_ptr;
+            if (!table_field->collected_stats)
+              continue;
+            if ((rc= table_field->collected_stats->add()))
+              break;
+          }
+          if (rc)
+            break;
+          rows++;
+        }
+      }
+      file->ha_rnd_end();
+    }
+    rc= (rc == HA_ERR_END_OF_FILE && !thd->killed) ? 0 : 1;
+  }
+
 
   /* 
     Calculate values for all statistical characteristics on columns and

storage/innobase/btr/btr0cur.cc

@@ -2849,14 +2849,18 @@ btr_cur_open_at_index_side(
 
 /**********************************************************************//**
 Positions a cursor at a randomly chosen position within a B-tree.
-@return true if the index is available and we have put the cursor, false
-if the index is unavailable */
-bool
+@return DB_SUCCESS if the index is available and we have put the cursor,
+error code if the index is unavailable. If simulate_uniform=true, could be
+DB_RECORD_NOT_FOUND returned, which means that no record is chosen in the
+generated tree path. The caller should retry a call, that will
+try a new tree path */
+dberr_t
 btr_cur_open_at_rnd_pos(
-	dict_index_t*	index,		/*!< in: index */
-	ulint		latch_mode,	/*!< in: BTR_SEARCH_LEAF, ... */
-	btr_cur_t*	cursor,		/*!< in/out: B-tree cursor */
-	mtr_t*		mtr)		/*!< in: mtr */
+	dict_index_t*	index,	    /*!< in: index */
+	ulint		latch_mode, /*!< in: BTR_SEARCH_LEAF, ... */
+	btr_cur_t*	cursor,	    /*!< in/out: B-tree cursor */
+	mtr_t*		mtr,        /*!< in: mtr */
+        bool sim_uniform_dist)      /*!< in: uniform distribution simulation */
 {
 	page_cur_t*	page_cursor;
 	ulint		node_ptr_max_size = srv_page_size / 2;
@@ -2868,6 +2872,11 @@ btr_cur_open_at_rnd_pos(
 	ulint		n_blocks = 0;
 	ulint		n_releases = 0;
 	mem_heap_t*	heap		= NULL;
+	/* For uniform distribution simulation is used naive A/R sampling
+	    algorithm, check link below for details
+	   https://courses.cs.washington.edu/courses/cse590q/05au/papers/Olken-Sampling.pdf
+	*/
+        double          p = 1.0;
 	rec_offs	offsets_[REC_OFFS_NORMAL_SIZE];
 	rec_offs*	offsets		= offsets_;
 	rec_offs_init(offsets_);
@@ -2917,15 +2926,15 @@ btr_cur_open_at_rnd_pos(
 	}
 
 	DBUG_EXECUTE_IF("test_index_is_unavailable",
-			return(false););
+			return(DB_ERROR););
 
 	if (index->page == FIL_NULL) {
 		/* Since we don't hold index lock until just now, the index
 		could be modified by others, for example, if this is a
 		statistics updater for referenced table, it could be marked
 		as unavailable by 'DROP TABLE' in the mean time, since
 		we don't hold lock for statistics updater */
-		return(false);
+		return(DB_ERROR);
 	}
 
 	const rw_lock_type_t root_leaf_rw_latch = btr_cur_latch_for_root_leaf(
@@ -3003,8 +3012,18 @@ btr_cur_open_at_rnd_pos(
 			/* We are in the root node */
 
 			height = btr_page_get_level(page);
+		} else {
+			if(sim_uniform_dist) {
+				ulint n_recs = page_get_n_recs(block->page.frame);
+				int extra_bytes = dict_table_is_comp(index->table) ?
+				REC_N_NEW_EXTRA_BYTES : REC_N_OLD_EXTRA_BYTES;
+				// Get max fanout from n_recs / (srv_page_size /
+				// extra_bytes +1), but exchange two divisons by
+				// one multiplication and one division
+				p *= (double)n_recs * (extra_bytes + 1) /
+					(double)srv_page_size;
+			}
 		}
-
 		if (height == 0) {
 			if (rw_latch == RW_NO_LATCH
 			    || srv_read_only_mode) {
@@ -3038,7 +3057,7 @@ btr_cur_open_at_rnd_pos(
 			}
 		}
 
-		page_cur_open_on_rnd_user_rec(block, page_cursor);
+                page_cur_open_on_rnd_user_rec(block, page_cursor);
 
 		if (height == 0) {
 
@@ -3133,7 +3152,14 @@ btr_cur_open_at_rnd_pos(
 		mem_heap_free(heap);
 	}
 
-	return err == DB_SUCCESS;
+	// We get max value of type by using ~(type)0 for
+	// getting 0..1 pseudo random number from ut_rnd_gen()
+	// and exchange division by multiplication like
+	// (b / c) < a <=> b < (a * c)
+	if(sim_uniform_dist && (ut_rnd_gen() < (p * ~(uint32_t)0)))
+		err = DB_RECORD_NOT_FOUND;
+
+	return err;
 }
 
 /*==================== B-TREE INSERT =========================*/

storage/innobase/dict/dict0stats.cc

@@ -1224,12 +1224,12 @@ btr_estimate_number_of_different_key_vals(dict_index_t* index,
 	for (i = 0; i < n_sample_pages; i++) {
 		mtr.start();
 
-		bool	available;
+                dberr_t	available;
 
 		available = btr_cur_open_at_rnd_pos(index, BTR_SEARCH_LEAF,
-						    &cursor, &mtr);
+						    &cursor, &mtr, false);
 
-		if (!available || index->table->bulk_trx_id != bulk_trx_id) {
+		if (available != DB_SUCCESS || index->table->bulk_trx_id != bulk_trx_id) {
 			mtr.commit();
 			mem_heap_free(heap);
 

storage/innobase/handler/ha_innodb.cc

@@ -113,6 +113,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include "fil0pagecompress.h"
 #include "ut0mem.h"
 #include "row0ext.h"
+#include "row0sel.h"
 
 #include "lz4.h"
 #include "lzo/lzo1x.h"
@@ -3020,7 +3021,8 @@ ha_innobase::ha_innobase(
                           | HA_CAN_ONLINE_BACKUPS
 			  | HA_CONCURRENT_OPTIMIZE
 			  | HA_CAN_SKIP_LOCKED
-			  |  (srv_force_primary_key ? HA_REQUIRE_PRIMARY_KEY : 0)
+			  |  (srv_force_primary_key ? HA_REQUIRE_PRIMARY_KEY : 0
+                          | HA_NATIVE_SAMPLING)
 		  ),
 	m_start_of_scan(),
         m_mysql_has_locked()
@@ -9489,6 +9491,66 @@ ha_innobase::rnd_next(
 	DBUG_RETURN(error);
 }
 
+int
+ha_innobase::sample_init()
+{
+  int		err;
+
+  /* Store the active index value so that we can restore the original
+  value after a scan */
+
+  if (m_prebuilt->clust_index_was_generated) {
+    err = change_active_index(MAX_KEY);
+  } else {
+    err = change_active_index(m_primary_key);
+  }
+
+  return(err);
+}
+
+int
+ha_innobase::sample_end()
+{
+  return(index_end());
+}
+int
+ha_innobase::sample_next(
+/*=====================*/
+    uchar *buf)
+{
+  mtr_t		mtr;
+  btr_pcur_t*	pcur= m_prebuilt->pcur;
+  rec_t*        rec;
+  bool          res;
+  dberr_t       err;
+  rec_offs	offsets_[REC_OFFS_NORMAL_SIZE];
+  rec_offs*	offsets= offsets_;
+  rec_offs_init(offsets_);
+
+  dict_index_t*	index= innobase_get_index(MAX_KEY);
+  mtr.start();
+  err= btr_pcur_open_at_rnd_pos(index, BTR_SEARCH_LEAF, pcur, &mtr, true);
+  mtr.commit();
+  if(err != DB_SUCCESS)
+  {
+    return HA_ERR_KEY_NOT_FOUND;
+  }
+  rec= btr_pcur_get_rec(pcur);
+  mem_heap_t*	heap= NULL;
+  auto _ = make_scope_exit([heap]() { if(heap) mem_heap_free(heap); });
+
+  offsets= rec_get_offsets(rec, index, offsets, index->n_core_fields, ULINT_UNDEFINED, &heap);
+  ut_ad(offsets);
+  if (!offsets)
+    return HA_ERR_INTERNAL_ERROR;
+
+  res= row_sel_store_mysql_rec(
+      buf, m_prebuilt, rec, NULL, true,
+      index, offsets);
+
+  return res ? 0 : HA_ERR_INTERNAL_ERROR;
+}
+
 /**********************************************************************//**
 Fetches a row from the table based on a row reference.
 @return 0, HA_ERR_KEY_NOT_FOUND, or error code */