MDEV-9764: MariaDB does not limit memory used for range optimization

A partial backport of 67f21fb3a077dedfd14b9ca720e926c55e682f93,
Bug#22283790: RANGE OPTIMIZER UTILIZES TOO MUCH MEMORY WITH MANY OR CONDITIONS

The backported part changes SEL_TREE::keys from being an array of
MAX_KEY elements (64*8=512 bytes) to a Mem_root_array<SEL_ARG*> (32 bytes +
alloc'ed array of as many elements as we need).

The patch doesn't fix the "not limiting memory" part, but the memory usage
is much lower with it.

2 files changed, 116 insertions, 38 deletions

diff --git a/sql/mem_root_array.h b/sql/mem_root_array.h
index 2dcc475cd7b..5daeedadcba 100644
--- a/sql/mem_root_array.h
+++ b/sql/mem_root_array.h
@@ -47,12 +47,21 @@ template<typename Element_type, bool has_trivial_destructor>
 class Mem_root_array
 {
 public:
+  /// Convenience typedef, same typedef name as std::vector
+  typedef Element_type value_type;
+
   Mem_root_array(MEM_ROOT *root)
     : m_root(root), m_array(NULL), m_size(0), m_capacity(0)
   {
     DBUG_ASSERT(m_root != NULL);
   }
 
+  Mem_root_array(MEM_ROOT *root, size_t n, const value_type &val= value_type())
+    : m_root(root), m_array(NULL), m_size(0), m_capacity(0)
+  {
+    resize(n, val);
+  }
+
   ~Mem_root_array()
   {
     clear();
@@ -70,6 +79,12 @@ public:
     return m_array[n];
   }
 
+  Element_type &operator[](size_t n) { return at(n); }
+  const Element_type &operator[](size_t n) const { return at(n); }
+
+  Element_type &back() { return at(size() - 1); }
+  const Element_type &back() const { return at(size() - 1); }
+
   // Returns a pointer to the first element in the array.
   Element_type *begin() { return &m_array[0]; }
 
@@ -155,6 +170,58 @@ public:
     return false;
   }
 
+  /**
+    Removes the last element in the array, effectively reducing the
+    container size by one. This destroys the removed element.
+   */
+  void pop_back()
+  {
+    DBUG_ASSERT(!empty());
+    if (!has_trivial_destructor)
+      back().~Element_type();
+    m_size-= 1;
+  }
+
+  /**
+    Resizes the container so that it contains n elements.
+
+    If n is smaller than the current container size, the content is
+    reduced to its first n elements, removing those beyond (and
+    destroying them).
+
+    If n is greater than the current container size, the content is
+    expanded by inserting at the end as many elements as needed to
+    reach a size of n. If val is specified, the new elements are
+    initialized as copies of val, otherwise, they are
+    value-initialized.
+
+    If n is also greater than the current container capacity, an automatic
+    reallocation of the allocated storage space takes place.
+
+    Notice that this function changes the actual content of the
+    container by inserting or erasing elements from it.
+   */
+  void resize(size_t n, const value_type &val= value_type())
+  {
+    if (n == m_size)
+      return;
+    if (n > m_size)
+    {
+      if (!reserve(n))
+      {
+        while (n != m_size)
+          push_back(val);
+      }
+      return;
+    }
+    if (!has_trivial_destructor)
+    {
+      while (n != m_size)
+        pop_back();
+    }
+    m_size= n;
+  }
+
   size_t capacity()     const { return m_capacity; }
   size_t element_size() const { return sizeof(Element_type); }
   bool   empty()        const { return size() == 0; }
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index ea5f77eed1d..0c8166c13f8 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -256,12 +256,19 @@ public:
        (type == SEL_TREE::IMPOSSIBLE)
   */
   enum Type { IMPOSSIBLE, ALWAYS, MAYBE, KEY, KEY_SMALLER } type;
-  SEL_TREE(enum Type type_arg) :type(type_arg) {}
-  SEL_TREE() :type(KEY)
+
+  SEL_TREE(enum Type type_arg, MEM_ROOT *root, size_t num_keys)
+    : type(type_arg), keys(root, num_keys), n_ror_scans(0)
   {
     keys_map.clear_all();
-    bzero((char*) keys,sizeof(keys));
   }
+
+  SEL_TREE(MEM_ROOT *root, size_t num_keys) :
+    type(KEY), keys(root, num_keys), n_ror_scans(0)
+  { 
+    keys_map.clear_all();
+  }
+   
   SEL_TREE(SEL_TREE *arg, bool without_merges, RANGE_OPT_PARAM *param);
   /*
     Note: there may exist SEL_TREE objects with sel_tree->type=KEY and
@@ -269,7 +276,8 @@ public:
     merit in range analyzer functions (e.g. get_mm_parts) returning a
     pointer to such SEL_TREE instead of NULL)
   */
-  SEL_ARG *keys[MAX_KEY];
+  Mem_root_array<SEL_ARG *, true> keys;
+
   key_map keys_map;        /* bitmask of non-NULL elements in keys */
 
   /*
@@ -579,7 +587,7 @@ int SEL_IMERGE::and_sel_tree(RANGE_OPT_PARAM *param, SEL_TREE *tree,
   {
     SEL_TREE *res_or_tree= 0;
     SEL_TREE *and_tree= 0;
-    if (!(res_or_tree= new SEL_TREE()) ||
+    if (!(res_or_tree= new SEL_TREE(param->mem_root, param->keys)) ||
         !(and_tree= new SEL_TREE(tree, TRUE, param)))
       return (-1);
     if (!and_range_trees(param, *or_tree, and_tree, res_or_tree))
@@ -788,7 +796,10 @@ int SEL_IMERGE::or_sel_imerge_with_checks(RANGE_OPT_PARAM *param,
 */ 
 
 SEL_TREE::SEL_TREE(SEL_TREE *arg, bool without_merges,
-                   RANGE_OPT_PARAM *param): Sql_alloc()
+                   RANGE_OPT_PARAM *param) 
+  : Sql_alloc(),
+    keys(param->mem_root, param->keys),
+    n_ror_scans(0)
 {
   keys_map= arg->keys_map;
   type= arg->type;
@@ -3020,9 +3031,7 @@ bool calculate_cond_selectivity_for_table(THD *thd, TABLE *table, Item **cond)
     PARAM param;
     MEM_ROOT alloc;
     SEL_TREE *tree;
-    SEL_ARG **key, **end;
     double rows;
-    uint idx= 0;
   
     init_sql_alloc(&alloc, thd->variables.range_alloc_block_size, 0,
                    MYF(MY_THREAD_SPECIFIC));
@@ -3067,11 +3076,12 @@ bool calculate_cond_selectivity_for_table(THD *thd, TABLE *table, Item **cond)
       goto free_alloc;
     }        
 
-    for (key= tree->keys, end= key + param.keys; key != end; key++, idx++)
+    for (uint idx= 0; idx < param.keys; idx++)
     {
-      if (*key)
+      SEL_ARG *key= tree->keys[idx];
+      if (key)
       {
-        if ((*key)->type == SEL_ARG::IMPOSSIBLE)
+        if (key->type == SEL_ARG::IMPOSSIBLE)
 	{
           rows= 0;
           table->reginfo.impossible_range= 1;
@@ -3079,10 +3089,10 @@ bool calculate_cond_selectivity_for_table(THD *thd, TABLE *table, Item **cond)
         }          
         else
         {
-          rows= records_in_column_ranges(&param, idx, *key);
+          rows= records_in_column_ranges(&param, idx, key);
           if (rows != HA_POS_ERROR)
-            (*key)->field->cond_selectivity= rows/table_records;
-        } 
+            key->field->cond_selectivity= rows/table_records;
+        }
       }
     }
 
@@ -4947,8 +4957,8 @@ TABLE_READ_PLAN *merge_same_index_scans(PARAM *param, SEL_IMERGE *imerge,
     {
       SEL_TREE **changed_tree= imerge->trees+(*tree_idx_ptr-1);
       SEL_ARG *key= (*changed_tree)->keys[key_idx];
-      bzero((*changed_tree)->keys,
-            sizeof((*changed_tree)->keys[0])*param->keys);
+      for (uint i= 0; i < param->keys; i++)
+        (*changed_tree)->keys[i]= NULL;
       (*changed_tree)->keys_map.clear_all();
       if (key) 
         key->incr_refs(); 
@@ -6725,8 +6735,8 @@ static TRP_RANGE *get_key_scans_params(PARAM *param, SEL_TREE *tree,
                                        bool update_tbl_stats,
                                        double read_time)
 {
-  uint idx;
-  SEL_ARG **key,**end, **key_to_read= NULL;
+  uint idx, best_idx;
+  SEL_ARG *key_to_read= NULL;
   ha_rows UNINIT_VAR(best_records);              /* protected by key_to_read */
   uint    UNINIT_VAR(best_mrr_flags),            /* protected by key_to_read */
           UNINIT_VAR(best_buf_size);             /* protected by key_to_read */
@@ -6749,9 +6759,10 @@ static TRP_RANGE *get_key_scans_params(PARAM *param, SEL_TREE *tree,
                                       sizeof(INDEX_SCAN_INFO *) * param->keys);
   }
   tree->index_scans_end= tree->index_scans;                                                  
-  for (idx= 0,key=tree->keys, end=key+param->keys; key != end; key++,idx++)
+  for (idx= 0; idx < param->keys; idx++)
   {
-    if (*key)
+    SEL_ARG *key= tree->keys[idx];
+    if (key)
     {
       ha_rows found_records;
       Cost_estimate cost;
@@ -6759,14 +6770,14 @@ static TRP_RANGE *get_key_scans_params(PARAM *param, SEL_TREE *tree,
       uint mrr_flags, buf_size;
       INDEX_SCAN_INFO *index_scan;
       uint keynr= param->real_keynr[idx];
-      if ((*key)->type == SEL_ARG::MAYBE_KEY ||
-          (*key)->maybe_flag)
+      if (key->type == SEL_ARG::MAYBE_KEY ||
+          key->maybe_flag)
         param->needed_reg->set_bit(keynr);
 
       bool read_index_only= index_read_must_be_used ? TRUE :
                             (bool) param->table->covering_keys.is_set(keynr);
 
-      found_records= check_quick_select(param, idx, read_index_only, *key,
+      found_records= check_quick_select(param, idx, read_index_only, key,
                                         update_tbl_stats, &mrr_flags,
                                         &buf_size, &cost);
 
@@ -6780,7 +6791,7 @@ static TRP_RANGE *get_key_scans_params(PARAM *param, SEL_TREE *tree,
         index_scan->used_key_parts= param->max_key_part+1;
         index_scan->range_count= param->range_count;
         index_scan->records= found_records;
-        index_scan->sel_arg= *key;
+        index_scan->sel_arg= key;
         *tree->index_scans_end++= index_scan;
       }        
       if ((found_records != HA_POS_ERROR) && param->is_ror_scan)
@@ -6794,6 +6805,7 @@ static TRP_RANGE *get_key_scans_params(PARAM *param, SEL_TREE *tree,
         read_time=    found_read_time;
         best_records= found_records;
         key_to_read=  key;
+        best_idx= idx;
         best_mrr_flags= mrr_flags;
         best_buf_size=  buf_size;
       }
@@ -6804,17 +6816,16 @@ static TRP_RANGE *get_key_scans_params(PARAM *param, SEL_TREE *tree,
                                       "ROR scans"););
   if (key_to_read)
   {
-    idx= key_to_read - tree->keys;
-    if ((read_plan= new (param->mem_root) TRP_RANGE(*key_to_read, idx,
+    if ((read_plan= new (param->mem_root) TRP_RANGE(key_to_read, best_idx,
                                                     best_mrr_flags)))
     {
       read_plan->records= best_records;
-      read_plan->is_ror= tree->ror_scans_map.is_set(idx);
+      read_plan->is_ror= tree->ror_scans_map.is_set(best_idx);
       read_plan->read_cost= read_time;
       read_plan->mrr_buf_size= best_buf_size;
       DBUG_PRINT("info",
                  ("Returning range plan for key %s, cost %g, records %lu",
-                  param->table->key_info[param->real_keynr[idx]].name,
+                  param->table->key_info[param->real_keynr[best_idx]].name,
                   read_plan->read_cost, (ulong) read_plan->records));
     }
   }
@@ -7442,9 +7453,11 @@ SEL_TREE *Item::get_mm_tree_for_const(RANGE_OPT_PARAM *param)
   MEM_ROOT *tmp_root= param->mem_root;
   param->thd->mem_root= param->old_root;
   SEL_TREE *tree;
-  tree= val_int() ? new(tmp_root) SEL_TREE(SEL_TREE::ALWAYS) :
-                    new(tmp_root) SEL_TREE(SEL_TREE::IMPOSSIBLE);
+
+  const SEL_TREE::Type type= val_int()? SEL_TREE::ALWAYS: SEL_TREE::IMPOSSIBLE;
   param->thd->mem_root= tmp_root;
+
+  tree= new (tmp_root) SEL_TREE(type, tmp_root, param->keys);
   DBUG_RETURN(tree);
 }
 
@@ -7470,7 +7483,8 @@ SEL_TREE *Item::get_mm_tree(RANGE_OPT_PARAM *param, Item **cond_ptr)
   if ((ref_tables & param->current_table) ||
       (ref_tables & ~(param->prev_tables | param->read_tables)))
     DBUG_RETURN(0);
-  DBUG_RETURN(new SEL_TREE(SEL_TREE::MAYBE));
+  DBUG_RETURN(new (param->mem_root) SEL_TREE(SEL_TREE::MAYBE, param->mem_root, 
+                                             param->keys));
 }
 
 
@@ -7588,7 +7602,8 @@ Item_bool_func::get_mm_parts(RANGE_OPT_PARAM *param, Field *field,
     if (field->eq(key_part->field))
     {
       SEL_ARG *sel_arg=0;
-      if (!tree && !(tree=new (param->thd->mem_root) SEL_TREE()))
+      if (!tree && !(tree=new (param->thd->mem_root) SEL_TREE(param->mem_root,
+                                                              param->keys)))
 	DBUG_RETURN(0);				// OOM
       if (!value || !(value_used_tables & ~param->read_tables))
       {
@@ -8551,7 +8566,7 @@ tree_or(RANGE_OPT_PARAM *param,SEL_TREE *tree1,SEL_TREE *tree2)
     imerge[0]= new SEL_IMERGE(tree1->merges.head(), 0, param);
   }
   bool no_imerge_from_ranges= FALSE;
-  if (!(result= new SEL_TREE()))
+  if (!(result= new (param->mem_root) SEL_TREE(param->mem_root, param->keys)))
     DBUG_RETURN(result);
 
   /* Build the range part of the tree for the formula (1) */ 
@@ -14382,16 +14397,12 @@ void QUICK_GROUP_MIN_MAX_SELECT::add_keys_and_lengths(String *key_names,
 static void print_sel_tree(PARAM *param, SEL_TREE *tree, key_map *tree_map,
                            const char *msg)
 {
-  SEL_ARG **key,**end;
-  int idx;
   char buff[1024];
   DBUG_ENTER("print_sel_tree");
 
   String tmp(buff,sizeof(buff),&my_charset_bin);
   tmp.length(0);
-  for (idx= 0,key=tree->keys, end=key+param->keys ;
-       key != end ;
-       key++,idx++)
+  for (uint idx= 0; idx < param->keys; idx++)
   {
     if (tree_map->is_set(idx))
     {