MDEV-26996 Support descending indexes in the range optimizer

Make the Range Optimizer support descending index key parts.

We follow the approach taken in MySQL-8.

See HowRangeOptimizerHandlesDescKeyparts for the description.

1 files changed, 113 insertions, 49 deletions

diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 1fbaec31636..959dedfb90a 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1879,6 +1879,7 @@ SEL_ARG::SEL_ARG(SEL_ARG &arg) :Sql_alloc()
   max_flag=arg.max_flag;
   maybe_flag=arg.maybe_flag;
   maybe_null=arg.maybe_null;
+  is_ascending= arg.is_ascending;
   part=arg.part;
   field=arg.field;
   min_value=arg.min_value;
@@ -1904,9 +1905,10 @@ inline void SEL_ARG::make_root()
   use_count=0; elements=1;
 }
 
-SEL_ARG::SEL_ARG(Field *f,const uchar *min_value_arg,
+SEL_ARG::SEL_ARG(Field *f, bool is_asc, const uchar *min_value_arg,
                  const uchar *max_value_arg)
   :min_flag(0), max_flag(0), maybe_flag(0), maybe_null(f->real_maybe_null()),
+   is_ascending(is_asc),
    elements(1), use_count(1), field(f), min_value((uchar*) min_value_arg),
    max_value((uchar*) max_value_arg), next(0),prev(0),
    next_key_part(0), color(BLACK), type(KEY_RANGE), weight(1)
@@ -1915,11 +1917,12 @@ SEL_ARG::SEL_ARG(Field *f,const uchar *min_value_arg,
   max_part_no= 1;
 }
 
-SEL_ARG::SEL_ARG(Field *field_,uint8 part_,
+SEL_ARG::SEL_ARG(Field *field_,uint8 part_, bool is_asc_,
                  uchar *min_value_, uchar *max_value_,
 		 uint8 min_flag_,uint8 max_flag_,uint8 maybe_flag_)
   :min_flag(min_flag_),max_flag(max_flag_),maybe_flag(maybe_flag_),
-   part(part_),maybe_null(field_->real_maybe_null()), elements(1),use_count(1),
+   part(part_),maybe_null(field_->real_maybe_null()), is_ascending(is_asc_),
+   elements(1),use_count(1),
    field(field_), min_value(min_value_), max_value(max_value_),
    next(0),prev(0),next_key_part(0),color(BLACK),type(KEY_RANGE), weight(1)
 {
@@ -1938,8 +1941,8 @@ SEL_ARG::SEL_ARG(Field *field_,uint8 part_,
 class SEL_ARG_LE: public SEL_ARG
 {
 public:
-  SEL_ARG_LE(const uchar *key, Field *field)
-   :SEL_ARG(field, key, key)
+  SEL_ARG_LE(const uchar *key, Field *field, bool is_asc)
+   :SEL_ARG(field, is_asc, key, key)
   {
     if (!field->real_maybe_null())
       min_flag= NO_MIN_RANGE;     // From start
@@ -1959,16 +1962,17 @@ public:
     Use this constructor if value->save_in_field() went precisely,
     without any data rounding or truncation.
   */
-  SEL_ARG_LT(const uchar *key, Field *field)
-   :SEL_ARG_LE(key, field)
+  SEL_ARG_LT(const uchar *key, Field *field, bool is_asc)
+   :SEL_ARG_LE(key, field, is_asc)
   { max_flag= NEAR_MAX; }
   /*
     Use this constructor if value->save_in_field() returned success,
     but we don't know if rounding or truncation happened
     (as some Field::store() do not report minor data changes).
   */
-  SEL_ARG_LT(THD *thd, const uchar *key, Field *field, Item *value)
-   :SEL_ARG_LE(key, field)
+  SEL_ARG_LT(THD *thd, const uchar *key, Field *field, bool is_asc,
+             Item *value)
+   :SEL_ARG_LE(key, field, is_asc)
   {
     if (stored_field_cmp_to_item(thd, field, value) == 0)
       max_flag= NEAR_MAX;
@@ -1984,7 +1988,7 @@ public:
     without any data rounding or truncation.
   */
   SEL_ARG_GT(const uchar *key, const KEY_PART *key_part, Field *field)
-   :SEL_ARG(field, key, key)
+   :SEL_ARG(field, !(key_part->flag & HA_REVERSE_SORT), key, key)
   {
     // Don't use open ranges for partial key_segments
     if (!(key_part->flag & HA_PART_KEY_SEG))
@@ -1998,7 +2002,7 @@ public:
   */
   SEL_ARG_GT(THD *thd, const uchar *key,
              const KEY_PART *key_part, Field *field, Item *value)
-   :SEL_ARG(field, key, key)
+   :SEL_ARG(field, !(key_part->flag & HA_REVERSE_SORT), key, key)
   {
     // Don't use open ranges for partial key_segments
     if ((!(key_part->flag & HA_PART_KEY_SEG)) &&
@@ -2016,8 +2020,8 @@ public:
     Use this constructor if value->save_in_field() went precisely,
     without any data rounding or truncation.
   */
-  SEL_ARG_GE(const uchar *key, Field *field)
-   :SEL_ARG(field, key, key)
+  SEL_ARG_GE(const uchar *key, Field *field, bool is_asc)
+   :SEL_ARG(field, is_asc, key, key)
   {
     max_flag= NO_MAX_RANGE;
   }
@@ -2028,7 +2032,7 @@ public:
   */
   SEL_ARG_GE(THD *thd, const uchar *key,
              const KEY_PART *key_part, Field *field, Item *value)
-   :SEL_ARG(field, key, key)
+   :SEL_ARG(field, !(key_part->flag & HA_REVERSE_SORT), key, key)
   {
     // Don't use open ranges for partial key_segments
     if ((!(key_part->flag & HA_PART_KEY_SEG)) &&
@@ -2059,7 +2063,8 @@ SEL_ARG *SEL_ARG::clone(RANGE_OPT_PARAM *param, SEL_ARG *new_parent,
   }
   else
   {
-    if (!(tmp= new (param->mem_root) SEL_ARG(field,part, min_value,max_value,
+    if (!(tmp= new (param->mem_root) SEL_ARG(field, part, is_ascending,
+                                             min_value, max_value,
                                              min_flag, max_flag, maybe_flag)))
       return 0;					// OOM
     tmp->parent=new_parent;
@@ -2830,6 +2835,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
       }
       trace_keypart.end();
       trace_idx_details.add("usable", !unusable_has_desc_keyparts);
+      unusable_has_desc_keyparts= false;
       if (unusable_has_desc_keyparts) // TODO MDEV-13756
       {
         key_parts= param.key[param.keys];
@@ -4423,12 +4429,14 @@ int find_used_partitions(PART_PRUNE_PARAM *ppar, SEL_ARG *key_tree)
           key_tree->next_key_part->store_min_key(ppar->key,
                                                  &tmp_min_key,
                                                  &tmp_min_flag,
-                                                 ppar->last_part_partno);
+                                                 ppar->last_part_partno,
+                                                 true);
         if (!tmp_max_flag)
           key_tree->next_key_part->store_max_key(ppar->key,
                                                  &tmp_max_key,
                                                  &tmp_max_flag,
-                                                 ppar->last_part_partno);
+                                                 ppar->last_part_partno,
+                                                 false);
         flag= tmp_min_flag | tmp_max_flag;
       }
       else
@@ -8674,7 +8682,8 @@ Item_func_null_predicate::get_mm_leaf(RANGE_OPT_PARAM *param,
   if (!field->real_maybe_null())
     DBUG_RETURN(type == ISNULL_FUNC ? &null_element : NULL);
   SEL_ARG *tree;
-  if (!(tree= new (alloc) SEL_ARG(field, is_null_string, is_null_string)))
+  bool is_asc= !(key_part->flag & HA_REVERSE_SORT);
+  if (!(tree= new (alloc) SEL_ARG(field, is_asc, is_null_string, is_null_string)))
     DBUG_RETURN(0);
   if (type == Item_func::ISNOTNULL_FUNC)
   {
@@ -8774,7 +8783,8 @@ Item_func_like::get_mm_leaf(RANGE_OPT_PARAM *param,
     int2store(min_str + maybe_null, min_length);
     int2store(max_str + maybe_null, max_length);
   }
-  SEL_ARG *tree= new (param->mem_root) SEL_ARG(field, min_str, max_str);
+  bool is_asc= !(key_part->flag & HA_REVERSE_SORT);
+  SEL_ARG *tree= new (param->mem_root) SEL_ARG(field, is_asc, min_str, max_str);
   DBUG_RETURN(tree);
 }
 
@@ -9022,18 +9032,19 @@ SEL_ARG *Field::stored_field_make_mm_leaf(RANGE_OPT_PARAM *param,
   if (!(str= make_key_image(param->mem_root, key_part)))
     DBUG_RETURN(0);
 
+  bool is_asc= !(key_part->flag & HA_REVERSE_SORT);
   switch (op) {
   case SCALAR_CMP_LE:
-    DBUG_RETURN(new (mem_root) SEL_ARG_LE(str, this));
+    DBUG_RETURN(new (mem_root) SEL_ARG_LE(str, this, is_asc));
   case SCALAR_CMP_LT:
-    DBUG_RETURN(new (mem_root) SEL_ARG_LT(thd, str, this, value));
+    DBUG_RETURN(new (mem_root) SEL_ARG_LT(thd, str, this, is_asc, value));
   case SCALAR_CMP_GT:
     DBUG_RETURN(new (mem_root) SEL_ARG_GT(thd, str, key_part, this, value));
   case SCALAR_CMP_GE:
     DBUG_RETURN(new (mem_root) SEL_ARG_GE(thd, str, key_part, this, value));
   case SCALAR_CMP_EQ:
   case SCALAR_CMP_EQUAL:
-    DBUG_RETURN(new (mem_root) SEL_ARG(this, str, str));
+    DBUG_RETURN(new (mem_root) SEL_ARG(this, is_asc, str, str));
     break;
   }
   DBUG_ASSERT(0);
@@ -9051,18 +9062,19 @@ SEL_ARG *Field::stored_field_make_mm_leaf_exact(RANGE_OPT_PARAM *param,
   if (!(str= make_key_image(param->mem_root, key_part)))
     DBUG_RETURN(0);
 
+  bool is_asc= !(key_part->flag & HA_REVERSE_SORT);
   switch (op) {
   case SCALAR_CMP_LE:
-    DBUG_RETURN(new (param->mem_root) SEL_ARG_LE(str, this));
+    DBUG_RETURN(new (param->mem_root) SEL_ARG_LE(str, this, is_asc));
   case SCALAR_CMP_LT:
-    DBUG_RETURN(new (param->mem_root) SEL_ARG_LT(str, this));
+    DBUG_RETURN(new (param->mem_root) SEL_ARG_LT(str, this, is_asc));
   case SCALAR_CMP_GT:
     DBUG_RETURN(new (param->mem_root) SEL_ARG_GT(str, key_part, this));
   case SCALAR_CMP_GE:
-    DBUG_RETURN(new (param->mem_root) SEL_ARG_GE(str, this));
+    DBUG_RETURN(new (param->mem_root) SEL_ARG_GE(str, this, is_asc));
   case SCALAR_CMP_EQ:
   case SCALAR_CMP_EQUAL:
-    DBUG_RETURN(new (param->mem_root) SEL_ARG(this, str, str));
+    DBUG_RETURN(new (param->mem_root) SEL_ARG(this, is_asc, str, str));
     break;
   }
   DBUG_ASSERT(0);
@@ -11780,6 +11792,46 @@ get_quick_select(PARAM *param,uint idx,SEL_ARG *key_tree, uint mrr_flags,
 }
 
 
+void SEL_ARG::store_next_min_max_keys(KEY_PART *key,
+                                      uchar **cur_min_key, uint *cur_min_flag,
+                                      uchar **cur_max_key, uint *cur_max_flag,
+                                      int *min_part, int *max_part)
+{
+  DBUG_ASSERT(next_key_part);
+  bool asc = next_key_part->is_ascending;
+
+  if (!get_min_flag())
+  {
+    if (asc)
+    {
+      *min_part += next_key_part->store_min_key(key, cur_min_key,
+                                                cur_min_flag, MAX_KEY, true);
+    }
+    else
+    {
+      uint tmp_flag = invert_min_flag(*cur_min_flag);
+      *min_part += next_key_part->store_max_key(key, cur_min_key, &tmp_flag,
+                                                MAX_KEY, true);
+      *cur_min_flag = invert_max_flag(tmp_flag);
+    }
+  }
+  if (!get_max_flag())
+  {
+    if (asc)
+    {
+      *max_part += next_key_part->store_max_key(key, cur_max_key,
+                                                cur_max_flag, MAX_KEY, false);
+    }
+    else
+    {
+      uint tmp_flag = invert_max_flag(*cur_max_flag);
+      *max_part += next_key_part->store_min_key(key, cur_max_key, &tmp_flag,
+                                                MAX_KEY, false);
+      *cur_max_flag = invert_min_flag(tmp_flag);
+    }
+  }
+}
+
 /*
 ** Fix this to get all possible sub_ranges
 */
@@ -11793,17 +11845,19 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
   int min_part= key_tree->part-1, // # of keypart values in min_key buffer
       max_part= key_tree->part-1; // # of keypart values in max_key buffer
 
-  if (key_tree->left != &null_element)
+  SEL_ARG *next_tree = key_tree->is_ascending ? key_tree->left : key_tree->right;
+  if (next_tree != &null_element)
   {
-    if (get_quick_keys(param,quick,key,key_tree->left,
+    if (get_quick_keys(param,quick,key,next_tree,
 		       min_key,min_key_flag, max_key, max_key_flag))
       return 1;
   }
   uchar *tmp_min_key=min_key,*tmp_max_key=max_key;
-  min_part+= key_tree->store_min(key[key_tree->part].store_length,
-                                 &tmp_min_key,min_key_flag);
-  max_part+= key_tree->store_max(key[key_tree->part].store_length,
-                                 &tmp_max_key,max_key_flag);
+
+  key_tree->store_min_max(key[key_tree->part].store_length,
+                          &tmp_min_key, min_key_flag,
+                          &tmp_max_key, max_key_flag,
+                          &min_part, &max_part);
 
   if (key_tree->next_key_part &&
       key_tree->next_key_part->type == SEL_ARG::KEY_RANGE &&
@@ -11813,31 +11867,40 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
          memcmp(min_key, max_key, (uint)(tmp_max_key - max_key))==0 &&
 	 key_tree->min_flag==0 && key_tree->max_flag==0)
     {
+      // psergey-note: simplified the parameters below as follows:
+      //  min_key_flag | key_tree->min_flag   ->   min_key_flag
+      //  max_key_flag | key_tree->max_flag   ->   max_key_flag
       if (get_quick_keys(param,quick,key,key_tree->next_key_part,
-			 tmp_min_key, min_key_flag | key_tree->min_flag,
-			 tmp_max_key, max_key_flag | key_tree->max_flag))
+			 tmp_min_key, min_key_flag,
+			 tmp_max_key, max_key_flag))
 	return 1;
       goto end;					// Ugly, but efficient
     }
     {
-      uint tmp_min_flag=key_tree->min_flag,tmp_max_flag=key_tree->max_flag;
-      if (!tmp_min_flag)
-        min_part+= key_tree->next_key_part->store_min_key(key,
-                                                          &tmp_min_key,
-                                                          &tmp_min_flag,
-                                                          MAX_KEY);
-      if (!tmp_max_flag)
-        max_part+= key_tree->next_key_part->store_max_key(key,
-                                                          &tmp_max_key,
-                                                          &tmp_max_flag,
-                                                          MAX_KEY);
+      uint tmp_min_flag= key_tree->get_min_flag();
+      uint tmp_max_flag= key_tree->get_max_flag();
+
+      key_tree->store_next_min_max_keys(key,
+                                        &tmp_min_key, &tmp_min_flag,
+                                        &tmp_max_key, &tmp_max_flag,
+                                        &min_part, &max_part);
       flag=tmp_min_flag | tmp_max_flag;
     }
   }
   else
   {
-    flag = (key_tree->min_flag & GEOM_FLAG) ?
-      key_tree->min_flag : key_tree->min_flag | key_tree->max_flag;
+    if (key_tree->is_ascending)
+    {
+      flag= (key_tree->min_flag & GEOM_FLAG) ? key_tree->min_flag:
+                                              (key_tree->min_flag |
+                                              key_tree->max_flag);
+    }
+    else
+    {
+      // Invert flags for DESC keypart
+      flag= invert_min_flag(key_tree->min_flag) |
+            invert_max_flag(key_tree->max_flag);
+    }
   }
 
   /*
@@ -11898,8 +11961,9 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
     return 1;
 
  end:
-  if (key_tree->right != &null_element)
-    return get_quick_keys(param,quick,key,key_tree->right,
+  next_tree = key_tree->is_ascending ? key_tree->right : key_tree->left;
+  if (next_tree != &null_element)
+    return get_quick_keys(param,quick,key,next_tree,
 			  min_key,min_key_flag,
 			  max_key,max_key_flag);
   return 0;