WL#3352, Introducing Column list partitioning, makes it possible to partition on most data types, makes it possible to prune on multi-field partitioning

17 files changed, 1770 insertions, 745 deletions

diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 8a81a759e2a..cc6558f2db0 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -19,7 +19,8 @@
 
 enum partition_keywords
 { 
-  PKW_HASH= 0, PKW_RANGE, PKW_LIST, PKW_KEY, PKW_MAXVALUE, PKW_LINEAR
+  PKW_HASH= 0, PKW_RANGE, PKW_LIST, PKW_KEY, PKW_MAXVALUE, PKW_LINEAR,
+  PKW_COLUMNS
 };
 
 /*
diff --git a/sql/item_create.cc b/sql/item_create.cc
index bf359b10caa..3adc0112ff8 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -2052,6 +2052,18 @@ protected:
   virtual ~Create_func_to_days() {}
 };
 
+class Create_func_to_seconds : public Create_func_arg1
+{
+public:
+  virtual Item* create(THD *thd, Item *arg1);
+
+  static Create_func_to_seconds s_singleton;
+
+protected:
+  Create_func_to_seconds() {}
+  virtual ~Create_func_to_seconds() {}
+};
+
 
 #ifdef HAVE_SPATIAL
 class Create_func_touches : public Create_func_arg2
@@ -4480,6 +4492,15 @@ Create_func_to_days::create(THD *thd, Item *arg1)
 }
 
 
+Create_func_to_seconds Create_func_to_seconds::s_singleton;
+
+Item*
+Create_func_to_seconds::create(THD *thd, Item *arg1)
+{
+  return new (thd->mem_root) Item_func_to_seconds(arg1);
+}
+
+
 #ifdef HAVE_SPATIAL
 Create_func_touches Create_func_touches::s_singleton;
 
@@ -4917,6 +4938,7 @@ static Native_func_registry func_array[] =
   { { C_STRING_WITH_LEN("TIME_TO_SEC") }, BUILDER(Create_func_time_to_sec)},
   { { C_STRING_WITH_LEN("TOUCHES") }, GEOM_BUILDER(Create_func_touches)},
   { { C_STRING_WITH_LEN("TO_DAYS") }, BUILDER(Create_func_to_days)},
+  { { C_STRING_WITH_LEN("TO_SECONDS") }, BUILDER(Create_func_to_seconds)},
   { { C_STRING_WITH_LEN("UCASE") }, BUILDER(Create_func_ucase)},
   { { C_STRING_WITH_LEN("UNCOMPRESS") }, BUILDER(Create_func_uncompress)},
   { { C_STRING_WITH_LEN("UNCOMPRESSED_LENGTH") }, BUILDER(Create_func_uncompressed_length)},
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index d79b0b02998..bad9b85b2b6 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -941,6 +941,27 @@ longlong Item_func_to_days::val_int()
 }
 
 
+longlong Item_func_to_seconds::val_int_endpoint(bool left_endp,
+                                                bool *incl_endp)
+{
+  longlong res= val_int();
+  return null_value ? LONGLONG_MIN : res;
+}
+
+longlong Item_func_to_seconds::val_int()
+{
+  DBUG_ASSERT(fixed == 1);
+  MYSQL_TIME ltime;
+  longlong seconds;
+  longlong days;
+  if (get_arg0_date(&ltime, TIME_NO_ZERO_DATE))
+    return 0;
+  seconds=ltime.hour*3600L+ltime.minute*60+ltime.second;
+  seconds=ltime.neg ? -seconds : seconds;
+  days= (longlong) calc_daynr(ltime.year,ltime.month,ltime.day);
+  return (seconds + days * (24L * 3600L));
+}
+
 /*
   Get information about this Item tree monotonicity
 
@@ -967,6 +988,18 @@ enum_monotonicity_info Item_func_to_days::get_monotonicity_info() const
   return NON_MONOTONIC;
 }
 
+enum_monotonicity_info Item_func_to_seconds::get_monotonicity_info() const
+{
+  if (args[0]->type() == Item::FIELD_ITEM)
+  {
+    if (args[0]->field_type() == MYSQL_TYPE_DATE)
+      return MONOTONIC_STRICT_INCREASING;
+    if (args[0]->field_type() == MYSQL_TYPE_DATETIME)
+      return MONOTONIC_INCREASING;
+  }
+  return NON_MONOTONIC;
+}
+
 
 longlong Item_func_to_days::val_int_endpoint(bool left_endp, bool *incl_endp)
 {
diff --git a/sql/item_timefunc.h b/sql/item_timefunc.h
index 9e3c2e8c89f..fd42ec307db 100644
--- a/sql/item_timefunc.h
+++ b/sql/item_timefunc.h
@@ -73,6 +73,24 @@ public:
 };
 
 
+class Item_func_to_seconds :public Item_int_func
+{
+public:
+  Item_func_to_seconds(Item *a) :Item_int_func(a) {}
+  longlong val_int();
+  const char *func_name() const { return "to_seconds"; }
+  void fix_length_and_dec() 
+  { 
+    decimals=0; 
+    max_length=6*MY_CHARSET_BIN_MB_MAXLEN;
+    maybe_null=1; 
+  }
+  enum_monotonicity_info get_monotonicity_info() const;
+  longlong val_int_endpoint(bool left_endp, bool *incl_endp);
+  bool check_partition_func_processor(uchar *bool_arg) { return FALSE;}
+};
+
+
 class Item_func_dayofmonth :public Item_int_func
 {
 public:
diff --git a/sql/lex.h b/sql/lex.h
index 0a85824f6f7..c42ece86993 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -113,6 +113,7 @@ static SYMBOL symbols[] = {
   { "COLLATION",	SYM(COLLATION_SYM)},
   { "COLUMN",		SYM(COLUMN_SYM)},
   { "COLUMNS",		SYM(COLUMNS)},
+  { "COLUMN_LIST",	SYM(COLUMN_LIST_SYM)},
   { "COMMENT",		SYM(COMMENT_SYM)},
   { "COMMIT",		SYM(COMMIT_SYM)},
   { "COMMITTED",	SYM(COMMITTED_SYM)},
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 47067c03a85..e226b51fc66 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -438,35 +438,55 @@ public:
     return 0;
   }
 
-  /* returns a number of keypart values appended to the key buffer */
-  int store_min_key(KEY_PART *key, uchar **range_key, uint *range_key_flag)
+  /*
+    Returns a number of keypart values appended to the key buffer
+    for min key and max key. This function is used by both Range
+    Analysis and Partition pruning. For partition pruning we have
+    to ensure that we don't store also subpartition fields. Thus
+    we have to stop at the last partition part and not step into
+    the subpartition fields. For Range Analysis we set last_part
+    to MAX_KEY which we should never reach.
+  */
+  int store_min_key(KEY_PART *key,
+                    uchar **range_key,
+                    uint *range_key_flag,
+                    uint last_part)
   {
     SEL_ARG *key_tree= first();
     uint res= key_tree->store_min(key[key_tree->part].store_length,
                                   range_key, *range_key_flag);
     *range_key_flag|= key_tree->min_flag;
     if (key_tree->next_key_part &&
+        key_tree->part != last_part &&
 	key_tree->next_key_part->part == key_tree->part+1 &&
 	!(*range_key_flag & (NO_MIN_RANGE | NEAR_MIN)) &&
 	key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
-      res+= key_tree->next_key_part->store_min_key(key, range_key,
-                                                   range_key_flag);
+      res+= key_tree->next_key_part->store_min_key(key,
+                                                   range_key,
+                                                   range_key_flag,
+                                                   last_part);
     return res;
   }
 
   /* returns a number of keypart values appended to the key buffer */
-  int store_max_key(KEY_PART *key, uchar **range_key, uint *range_key_flag)
+  int store_max_key(KEY_PART *key,
+                    uchar **range_key,
+                    uint *range_key_flag,
+                    uint last_part)
   {
     SEL_ARG *key_tree= last();
     uint res=key_tree->store_max(key[key_tree->part].store_length,
                                  range_key, *range_key_flag);
     (*range_key_flag)|= key_tree->max_flag;
     if (key_tree->next_key_part &&
+        key_tree->part != last_part &&
 	key_tree->next_key_part->part == key_tree->part+1 &&
 	!(*range_key_flag & (NO_MAX_RANGE | NEAR_MAX)) &&
 	key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
-      res+= key_tree->next_key_part->store_max_key(key, range_key,
-                                                   range_key_flag);
+      res+= key_tree->next_key_part->store_max_key(key,
+                                                   range_key,
+                                                   range_key_flag,
+                                                   last_part);
     return res;
   }
 
@@ -634,6 +654,12 @@ public:
     using_real_indexes==TRUE
   */
   uint real_keynr[MAX_KEY];
+
+  /* Used to store 'current key tuples', in both range analysis and
+   * partitioning (list) analysis*/
+  uchar min_key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH],
+    max_key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH];
+
   /* Number of SEL_ARG objects allocated by SEL_ARG::clone_tree operations */
   uint alloced_sel_args; 
 };
@@ -645,8 +671,6 @@ public:
   longlong baseflag;
   uint max_key_part, range_count;
 
-  uchar min_key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH],
-    max_key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH];
   bool quick;				// Don't calulate possible keys
 
   uint fields_bitmap_size;
@@ -2599,6 +2623,8 @@ typedef struct st_part_prune_param
   /* Same as above for subpartitioning */
   my_bool *is_subpart_keypart;
 
+  my_bool ignore_part_fields; /* Ignore rest of partioning fields */
+
   /***************************************************************
    Following fields form find_used_partitions() recursion context:
    **************************************************************/
@@ -2614,6 +2640,11 @@ typedef struct st_part_prune_param
 
   /* Initialized bitmap of no_subparts size */
   MY_BITMAP subparts_bitmap;
+
+  uchar *cur_min_key;
+  uchar *cur_max_key;
+
+  uint cur_min_flag, cur_max_flag;
 } PART_PRUNE_PARAM;
 
 static bool create_partition_index_description(PART_PRUNE_PARAM *prune_par);
@@ -2731,6 +2762,11 @@ bool prune_partitions(THD *thd, TABLE *table, Item *pprune_cond)
     prune_param.arg_stack_end= prune_param.arg_stack;
     prune_param.cur_part_fields= 0;
     prune_param.cur_subpart_fields= 0;
+    
+    prune_param.cur_min_key= prune_param.range_param.min_key;
+    prune_param.cur_max_key= prune_param.range_param.max_key;
+    prune_param.cur_min_flag= prune_param.cur_max_flag= 0;
+
     init_all_partitions_iterator(part_info, &prune_param.part_iter);
     if (!tree->keys[0] || (-1 == (res= find_used_partitions(&prune_param,
                                                             tree->keys[0]))))
@@ -2967,6 +3003,11 @@ int find_used_partitions_imerge(PART_PRUNE_PARAM *ppar, SEL_IMERGE *imerge)
     ppar->arg_stack_end= ppar->arg_stack;
     ppar->cur_part_fields= 0;
     ppar->cur_subpart_fields= 0;
+    
+    ppar->cur_min_key= ppar->range_param.min_key;
+    ppar->cur_max_key= ppar->range_param.max_key;
+    ppar->cur_min_flag= ppar->cur_max_flag= 0;
+
     init_all_partitions_iterator(ppar->part_info, &ppar->part_iter);
     SEL_ARG *key_tree= (*ptree)->keys[0];
     if (!key_tree || (-1 == (res |= find_used_partitions(ppar, key_tree))))
@@ -3091,8 +3132,12 @@ int find_used_partitions(PART_PRUNE_PARAM *ppar, SEL_ARG *key_tree)
 {
   int res, left_res=0, right_res=0;
   int partno= (int)key_tree->part;
-  bool pushed= FALSE;
   bool set_full_part_if_bad_ret= FALSE;
+  bool ignore_part_fields= ppar->ignore_part_fields;
+  bool did_set_ignore_part_fields= FALSE;
+
+  if (check_stack_overrun(ppar->range_param.thd, 3*STACK_MIN_SIZE, NULL))
+    return -1;
 
   if (key_tree->left != &null_element)
   {
@@ -3100,35 +3145,153 @@ int find_used_partitions(PART_PRUNE_PARAM *ppar, SEL_ARG *key_tree)
       return -1;
   }
 
+  /* Push SEL_ARG's to stack to enable looking backwards as well */
+  ppar->cur_part_fields+= ppar->is_part_keypart[partno];
+  ppar->cur_subpart_fields+= ppar->is_subpart_keypart[partno];
+  *(ppar->arg_stack_end++)= key_tree;
+
   if (key_tree->type == SEL_ARG::KEY_RANGE)
   {
-    if (partno == 0 && (NULL != ppar->part_info->get_part_iter_for_interval))
+    if (ppar->part_info->get_part_iter_for_interval && 
+        key_tree->part <= ppar->last_part_partno)
     {
-      /* 
-        Partitioning is done by RANGE|INTERVAL(monotonic_expr(fieldX)), and
-        we got "const1 CMP fieldX CMP const2" interval <-- psergey-todo: change
+      if (ignore_part_fields)
+      {
+        /*
+          We come here when a condition on the first partitioning
+          fields led to evaluating the partitioning condition
+          (due to finding a condition of the type a < const or
+          b > const). Thus we must ignore the rest of the
+          partitioning fields but we still want to analyse the
+          subpartitioning fields.
+        */
+        if (key_tree->next_key_part)
+          res= find_used_partitions(ppar, key_tree->next_key_part);
+        else
+          res= -1;
+        goto pop_and_go_right;
+      }
+      /* Collect left and right bound, their lengths and flags */
+      uchar *min_key= ppar->cur_min_key;
+      uchar *max_key= ppar->cur_max_key;
+      uchar *tmp_min_key= min_key;
+      uchar *tmp_max_key= max_key;
+      key_tree->store_min(ppar->key[key_tree->part].store_length,
+                          &tmp_min_key, ppar->cur_min_flag);
+      key_tree->store_max(ppar->key[key_tree->part].store_length,
+                          &tmp_max_key, ppar->cur_max_flag);
+      uint flag;
+      if (key_tree->next_key_part &&
+          key_tree->next_key_part->part == key_tree->part+1 &&
+          key_tree->next_key_part->part <= ppar->last_part_partno &&
+          key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
+      {
+        /*
+          There are more key parts for partition pruning to handle
+          This mainly happens when the condition is an equality
+          condition.
+        */
+        if ((tmp_min_key - min_key) == (tmp_max_key - max_key) && 
+            (memcmp(min_key, max_key, (uint)(tmp_max_key - max_key)) == 0) &&
+            !key_tree->min_flag && !key_tree->max_flag)
+        {
+          /* Set 'parameters' */
+          ppar->cur_min_key= tmp_min_key;
+          ppar->cur_max_key= tmp_max_key;
+          uint save_min_flag= ppar->cur_min_flag;
+          uint save_max_flag= ppar->cur_max_flag;
+
+          ppar->cur_min_flag|= key_tree->min_flag;
+          ppar->cur_max_flag|= key_tree->max_flag;
+          
+          res= find_used_partitions(ppar, key_tree->next_key_part);
+           
+          /* Restore 'parameters' back */
+          ppar->cur_min_key= min_key;
+          ppar->cur_max_key= max_key;
+
+          ppar->cur_min_flag= save_min_flag;
+          ppar->cur_max_flag= save_max_flag;
+          goto pop_and_go_right;
+        }
+        /* We have arrived at the last field in the partition pruning */
+        uint tmp_min_flag= key_tree->min_flag,
+             tmp_max_flag= key_tree->max_flag;
+        if (!tmp_min_flag)
+          key_tree->next_key_part->store_min_key(ppar->key,
+                                                 &tmp_min_key,
+                                                 &tmp_min_flag,
+                                                 ppar->last_part_partno);
+        if (!tmp_max_flag)
+          key_tree->next_key_part->store_max_key(ppar->key,
+                                                 &tmp_max_key,
+                                                 &tmp_max_flag,
+                                                 ppar->last_part_partno);
+        flag= tmp_min_flag | tmp_max_flag;
+      }
+      else
+        flag= key_tree->min_flag | key_tree->max_flag;
+      
+      if (tmp_min_key != ppar->range_param.min_key)
+        flag&= ~NO_MIN_RANGE;
+      else
+        flag|= NO_MIN_RANGE;
+      if (tmp_max_key != ppar->range_param.max_key)
+        flag&= ~NO_MAX_RANGE;
+      else
+        flag|= NO_MAX_RANGE;
+
+      /*
+        We need to call the interval mapper if we have a condition which
+        makes sense to prune on. In the example of a COLUMN_LIST on a and
+        b it makes sense if we have a condition on a, or conditions on
+        both a and b. If we only have conditions on b it might make sense
+        but this is a harder case we will solve later. For the harder case
+        this clause then turns into use of all partitions and thus we
+        simply set res= -1 as if the mapper had returned that.
       */
-      DBUG_EXECUTE("info", dbug_print_segment_range(key_tree,
-                                                    ppar->range_param.
-                                                    key_parts););
-      res= ppar->part_info->
-           get_part_iter_for_interval(ppar->part_info,
-                                      FALSE,
-                                      key_tree->min_value, 
-                                      key_tree->max_value,
-                                      key_tree->min_flag | key_tree->max_flag,
-                                      &ppar->part_iter);
-      if (!res)
-        goto go_right; /* res==0 --> no satisfying partitions */
+      if (ppar->arg_stack[0]->part == 0)
+      {
+        uint32 i;
+        uint32 store_length_array[MAX_KEY];
+        uint32 num_keys= ppar->part_fields;
+
+        for (i= 0; i < num_keys; i++)
+          store_length_array[i]= ppar->key[i].store_length;
+        res= ppar->part_info->
+             get_part_iter_for_interval(ppar->part_info,
+                                        FALSE,
+                                        store_length_array,
+                                        ppar->range_param.min_key,
+                                        ppar->range_param.max_key,
+                                    tmp_min_key - ppar->range_param.min_key,
+                                    tmp_max_key - ppar->range_param.max_key,
+                                        flag,
+                                        &ppar->part_iter);
+        if (!res)
+          goto pop_and_go_right; /* res==0 --> no satisfying partitions */
+      }
+      else
+        res= -1;
+
       if (res == -1)
       {
-        //get a full range iterator
+        /* get a full range iterator */
         init_all_partitions_iterator(ppar->part_info, &ppar->part_iter);
       }
       /* 
         Save our intent to mark full partition as used if we will not be able
         to obtain further limits on subpartitions
       */
+      if (partno < ppar->last_part_partno)
+      {
+        /*
+          We need to ignore the rest of the partitioning fields in all
+          evaluations after this
+        */
+        did_set_ignore_part_fields= TRUE;
+        ppar->ignore_part_fields= TRUE;
+      }
       set_full_part_if_bad_ret= TRUE;
       goto process_next_key_part;
     }
@@ -3143,13 +3306,16 @@ int find_used_partitions(PART_PRUNE_PARAM *ppar, SEL_ARG *key_tree)
       res= ppar->part_info->
            get_subpart_iter_for_interval(ppar->part_info,
                                          TRUE,
+                                         NULL, /* Currently not used here */
                                          key_tree->min_value, 
                                          key_tree->max_value,
-                                         key_tree->min_flag | key_tree->max_flag,
+                                         0, 0, /* Those are ignored here */
+                                         key_tree->min_flag |
+                                           key_tree->max_flag,
                                          &subpart_iter);
       DBUG_ASSERT(res); /* We can't get "no satisfying subpartitions" */
       if (res == -1)
-        return -1; /* all subpartitions satisfy */
+        goto pop_and_go_right; /* all subpartitions satisfy */
         
       uint32 subpart_id;
       bitmap_clear_all(&ppar->subparts_bitmap);
@@ -3167,18 +3333,14 @@ int find_used_partitions(PART_PRUNE_PARAM *ppar, SEL_ARG *key_tree)
             bitmap_set_bit(&ppar->part_info->used_partitions,
                            part_id * ppar->part_info->no_subparts + i);
       }
-      goto go_right;
+      goto pop_and_go_right;
     }
 
     if (key_tree->is_singlepoint())
     {
-      pushed= TRUE;
-      ppar->cur_part_fields+=    ppar->is_part_keypart[partno];
-      ppar->cur_subpart_fields+= ppar->is_subpart_keypart[partno];
-      *(ppar->arg_stack_end++) = key_tree;
-
       if (partno == ppar->last_part_partno &&
-          ppar->cur_part_fields == ppar->part_fields)
+          ppar->cur_part_fields == ppar->part_fields &&
+          ppar->part_info->get_part_iter_for_interval == NULL)
       {
         /* 
           Ok, we've got "fieldN<=>constN"-type SEL_ARGs for all partitioning
@@ -3245,7 +3407,10 @@ int find_used_partitions(PART_PRUNE_PARAM *ppar, SEL_ARG *key_tree)
         able to infer any suitable condition, so bail out.
       */
       if (partno >= ppar->last_part_partno)
-        return -1;
+      {
+        res= -1;
+        goto pop_and_go_right;
+      }
     }
   }
 
@@ -3254,7 +3419,17 @@ process_next_key_part:
     res= find_used_partitions(ppar, key_tree->next_key_part);
   else
     res= -1;
- 
+
+  if (did_set_ignore_part_fields)
+  {
+    /*
+      We have returned from processing all key trees linked to our next
+      key part. We are ready to be moving down (using right pointers) and
+      this tree is a new evaluation requiring its own decision on whether
+      to ignore partitioning fields.
+    */
+    ppar->ignore_part_fields= FALSE;
+  }
   if (set_full_part_if_bad_ret)
   {
     if (res == -1)
@@ -3277,18 +3452,14 @@ process_next_key_part:
     init_all_partitions_iterator(ppar->part_info, &ppar->part_iter);
   }
 
-  if (pushed)
-  {
 pop_and_go_right:
-    /* Pop this key part info off the "stack" */
-    ppar->arg_stack_end--;
-    ppar->cur_part_fields-=    ppar->is_part_keypart[partno];
-    ppar->cur_subpart_fields-= ppar->is_subpart_keypart[partno];
-  }
+  /* Pop this key part info off the "stack" */
+  ppar->arg_stack_end--;
+  ppar->cur_part_fields-=    ppar->is_part_keypart[partno];
+  ppar->cur_subpart_fields-= ppar->is_subpart_keypart[partno];
 
   if (res == -1)
     return -1;
-go_right:
   if (key_tree->right != &null_element)
   {
     if (-1 == (right_res= find_used_partitions(ppar,key_tree->right)))
@@ -3377,6 +3548,7 @@ static bool create_partition_index_description(PART_PRUNE_PARAM *ppar)
   
   uint total_parts= used_part_fields + used_subpart_fields;
 
+  ppar->ignore_part_fields= FALSE;
   ppar->part_fields=      used_part_fields;
   ppar->last_part_partno= (int)used_part_fields - 1;
 
@@ -7477,12 +7649,16 @@ check_quick_keys(PARAM *param, uint idx, SEL_ARG *key_tree,
     tmp_max_flag=key_tree->max_flag;
     if (!tmp_min_flag)
       tmp_min_keypart+=
-      key_tree->next_key_part->store_min_key(param->key[idx], &tmp_min_key,
-					     &tmp_min_flag);
+      key_tree->next_key_part->store_min_key(param->key[idx],
+                                             &tmp_min_key,
+					     &tmp_min_flag,
+                                             MAX_KEY);
     if (!tmp_max_flag)
       tmp_max_keypart+=
-      key_tree->next_key_part->store_max_key(param->key[idx], &tmp_max_key,
-					     &tmp_max_flag);
+      key_tree->next_key_part->store_max_key(param->key[idx],
+                                             &tmp_max_key,
+					     &tmp_max_flag,
+                                             MAX_KEY);
     min_key_length= (uint) (tmp_min_key - param->min_key);
     max_key_length= (uint) (tmp_max_key - param->max_key);
   }
@@ -7752,11 +7928,15 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
     {
       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);
+        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_part+= key_tree->next_key_part->store_max_key(key,
+                                                          &tmp_max_key,
+                                                          &tmp_max_flag,
+                                                          MAX_KEY);
       flag=tmp_min_flag | tmp_max_flag;
     }
   }
diff --git a/sql/partition_element.h b/sql/partition_element.h
index 905bc38165b..d749681fe9b 100644
--- a/sql/partition_element.h
+++ b/sql/partition_element.h
@@ -1,4 +1,4 @@
-/* Copyright (C) 2006 MySQL AB
+/* Copyright (C) 2006-2009 MySQL AB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -36,6 +36,35 @@ enum partition_state {
 };
 
 /*
+  This struct is used to keep track of column expressions as part
+  of the COLUMNS concept in conjunction with RANGE and LIST partitioning.
+  The value can be either of MINVALUE, MAXVALUE and an expression that
+  must be constant and evaluate to the same type as the column it
+  represents.
+
+  The data in this fixed in two steps. The parser will only fill in whether
+  it is a max_value or provide an expression. Filling in
+  column_value, part_info, partition_id, null_value is done by the
+  function fix_column_value_function. However the item tree needs
+  fixed also before writing it into the frm file (in add_column_list_values).
+  To distinguish between those two variants, fixed= 1 after the
+  fixing in add_column_list_values and fixed= 2 otherwise. This is
+  since the fixing in add_column_list_values isn't a complete fixing.
+*/
+
+typedef struct p_column_list_val
+{
+  void* column_value;
+  Item* item_expression;
+  partition_info *part_info;
+  uint partition_id;
+  bool max_value;
+  bool null_value;
+  char fixed;
+} part_column_list_val;
+
+
+/*
   This struct is used to contain the value of an element
   in the VALUES IN struct. It needs to keep knowledge of
   whether it is a signed/unsigned value and whether it is
@@ -47,6 +76,7 @@ typedef struct p_elem_val
   longlong value;
   bool null_value;
   bool unsigned_flag;
+  part_column_list_val *col_val_array;
 } part_elem_value;
 
 struct st_ddl_log_memory_entry;
@@ -68,8 +98,9 @@ public:
   enum partition_state part_state;
   uint16 nodegroup_id;
   bool has_null_value;
-  bool signed_flag;/* Indicate whether this partition uses signed constants */
-  bool max_value;  /* Indicate whether this partition uses MAXVALUE */
+  /* signed_flag and max_value only relevant for subpartitions */
+  bool signed_flag;
+  bool max_value;
 
   partition_element()
   : part_max_rows(0), part_min_rows(0), range_value(0),
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index e2027d3571e..bb7c7c2be0f 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -591,6 +591,7 @@ error:
 
   SYNOPSIS
     check_range_constants()
+    thd                          Thread object
 
   RETURN VALUE
     TRUE                An error occurred during creation of range constants
@@ -603,76 +604,112 @@ error:
     called for RANGE PARTITIONed tables.
 */
 
-bool partition_info::check_range_constants()
+bool partition_info::check_range_constants(THD *thd)
 {
   partition_element* part_def;
-  longlong current_largest;
-  longlong part_range_value;
   bool first= TRUE;
   uint i;
   List_iterator<partition_element> it(partitions);
-  bool result= TRUE;
-  bool signed_flag= !part_expr->unsigned_flag;
+  int result= TRUE;
   DBUG_ENTER("partition_info::check_range_constants");
-  DBUG_PRINT("enter", ("INT_RESULT with %d parts", no_parts));
-
-  LINT_INIT(current_largest);
+  DBUG_PRINT("enter", ("RANGE with %d parts, column_list = %u", no_parts,
+                                                         column_list));
 
-  part_result_type= INT_RESULT;
-  range_int_array= (longlong*)sql_alloc(no_parts * sizeof(longlong));
-  if (unlikely(range_int_array == NULL))
+  if (column_list)
   {
-    mem_alloc_error(no_parts * sizeof(longlong));
-    goto end;
-  }
-  i= 0;
-  do
-  {
-    part_def= it++;
-    if ((i != (no_parts - 1)) || !defined_max_value)
+    part_column_list_val* loc_range_col_array;
+    part_column_list_val *current_largest_col_val;
+    uint no_column_values= part_field_list.elements;
+    uint size_entries= sizeof(part_column_list_val) * no_column_values;
+    range_col_array= (part_column_list_val*)sql_calloc(no_parts *
+                                                       size_entries);
+    LINT_INIT(current_largest_col_val);
+    if (unlikely(range_col_array == NULL))
     {
-      part_range_value= part_def->range_value;
-      if (!signed_flag)
-        part_range_value-= 0x8000000000000000ULL;
+      mem_alloc_error(no_parts * sizeof(longlong));
+      goto end;
     }
-    else
-      part_range_value= LONGLONG_MAX;
-    if (first)
+    loc_range_col_array= range_col_array;
+    i= 0;
+    do
     {
-      current_largest= part_range_value;
-      range_int_array[0]= part_range_value;
+      part_def= it++;
+      {
+        List_iterator<part_elem_value> list_val_it(part_def->list_val_list);
+        part_elem_value *range_val= list_val_it++;
+        part_column_list_val *col_val= range_val->col_val_array;
+
+        if (fix_column_value_functions(thd, col_val, i))
+          goto end;
+        memcpy(loc_range_col_array, (const void*)col_val, size_entries);
+        loc_range_col_array+= no_column_values;
+        if (!first)
+        {
+          if (compare_column_values((const void*)current_largest_col_val,
+                                    (const void*)col_val) >= 0)
+            goto range_not_increasing_error;
+        }
+        current_largest_col_val= col_val;
+      }
       first= FALSE;
+    } while (++i < no_parts);
+  }
+  else
+  {
+    longlong current_largest;
+    longlong part_range_value;
+    bool signed_flag= !part_expr->unsigned_flag;
+
+    LINT_INIT(current_largest);
+
+    part_result_type= INT_RESULT;
+    range_int_array= (longlong*)sql_alloc(no_parts * sizeof(longlong));
+    if (unlikely(range_int_array == NULL))
+    {
+      mem_alloc_error(no_parts * sizeof(longlong));
+      goto end;
     }
-    else
+    i= 0;
+    do
     {
-      if (likely(current_largest < part_range_value))
+      part_def= it++;
+      if ((i != (no_parts - 1)) || !defined_max_value)
       {
-        current_largest= part_range_value;
-        range_int_array[i]= part_range_value;
-      }
-      else if (defined_max_value &&
-               current_largest == part_range_value &&
-               part_range_value == LONGLONG_MAX &&
-               i == (no_parts - 1))
-      {
-        range_int_array[i]= part_range_value;
+        part_range_value= part_def->range_value;
+        if (!signed_flag)
+          part_range_value-= 0x8000000000000000ULL;
       }
       else
+        part_range_value= LONGLONG_MAX;
+
+      if (!first)
       {
-        my_error(ER_RANGE_NOT_INCREASING_ERROR, MYF(0));
-        goto end;
+        if (unlikely(current_largest > part_range_value) ||
+            (unlikely(current_largest == part_range_value) &&
+            (part_range_value < LONGLONG_MAX ||
+             i != (no_parts - 1) ||
+             !defined_max_value)))
+          goto range_not_increasing_error;
       }
-    }
-  } while (++i < no_parts);
+      range_int_array[i]= part_range_value;
+      current_largest= part_range_value;
+      first= FALSE;
+    } while (++i < no_parts);
+  }
   result= FALSE;
 end:
   DBUG_RETURN(result);
+
+range_not_increasing_error:
+  my_error(ER_RANGE_NOT_INCREASING_ERROR, MYF(0));
+  goto end;
 }
 
 
 /*
   Support routines for check_list_constants used by qsort to sort the
-  constant list expressions. One routine for unsigned and one for signed.
+  constant list expressions. One routine for integers and one for
+  column lists.
 
   SYNOPSIS
     list_part_cmp()
@@ -697,6 +734,133 @@ int partition_info::list_part_cmp(const void* a, const void* b)
     return 0;
 }
 
+ /*
+  Compare two lists of column values in RANGE/LIST partitioning
+  SYNOPSIS
+    compare_column_values()
+    first                    First column list argument
+    second                   Second column list argument
+  RETURN VALUES
+    0                        Equal
+    -1                       First argument is smaller
+    +1                       First argument is larger
+*/
+
+int partition_info::compare_column_values(const void *first_arg,
+                                          const void *second_arg)
+{
+  const part_column_list_val *first= (part_column_list_val*)first_arg;
+  const part_column_list_val *second= (part_column_list_val*)second_arg;
+  partition_info *part_info= first->part_info;
+  Field **field;
+
+  for (field= part_info->part_field_array; *field;
+       field++, first++, second++)
+  {
+    if (first->max_value || second->max_value)
+    {
+      if (first->max_value && second->max_value)
+        continue;
+      if (second->max_value)
+        return -1;
+      else
+        return +1;
+    }
+    if (first->null_value || second->null_value)
+    {
+      if (first->null_value && second->null_value)
+        continue;
+      if (second->null_value)
+        return +1;
+      else
+        return -1;
+    }
+    int res= (*field)->cmp((const uchar*)first->column_value,
+                           (const uchar*)second->column_value);
+    if (res)
+      return res;
+  }
+  return 0;
+}
+
+/*
+  Evaluate VALUES functions for column list values
+  SYNOPSIS
+    fix_column_value_functions()
+    thd                              Thread object
+    col_val                          List of column values
+    part_id                          Partition id we are fixing
+  RETURN VALUES
+    TRUE                             Error
+    FALSE                            Success
+  DESCRIPTION
+    Fix column VALUES and store in memory array adapted to the data type
+*/
+
+bool partition_info::fix_column_value_functions(THD *thd,
+                                                part_column_list_val *col_val,
+                                                uint part_id)
+{
+  uint no_columns= part_field_list.elements;
+  Name_resolution_context *context= &thd->lex->current_select->context;
+  TABLE_LIST *save_list= context->table_list;
+  bool result= FALSE;
+  uint i;
+  const char *save_where= thd->where;
+  DBUG_ENTER("partition_info::fix_column_value_functions");
+  if (col_val->fixed > 1)
+  {
+    DBUG_RETURN(FALSE);
+  }
+  context->table_list= 0;
+  thd->where= "partition function";
+  for (i= 0; i < no_columns; col_val++, i++)
+  {
+    Item *column_item= col_val->item_expression;
+    Field *field= part_field_array[i];
+    col_val->part_info= this;
+    col_val->partition_id= part_id;
+    if (col_val->max_value)
+      col_val->column_value= NULL;
+    else
+    {
+      if (!col_val->fixed && 
+          (column_item->fix_fields(thd, (Item**)0) ||
+          (!column_item->const_item())))
+      {
+        my_error(ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR, MYF(0));
+        result= TRUE;
+        goto end;
+      }
+      col_val->null_value= column_item->null_value;
+      col_val->column_value= NULL;
+      if (!col_val->null_value)
+      {
+        uchar *val_ptr;
+        uint len= field->pack_length();
+        if (column_item->save_in_field(field, TRUE))
+        {
+          my_error(ER_WRONG_TYPE_COLUMN_VALUE_ERROR, MYF(0));
+          result= TRUE;
+          goto end;
+        }
+        if (!(val_ptr= (uchar*) sql_calloc(len)))
+        {
+          mem_alloc_error(len);
+          result= TRUE;
+          goto end;
+        }
+        col_val->column_value= val_ptr;
+        memcpy(val_ptr, field->ptr, len);
+      }
+    }
+    col_val->fixed= 2;
+  }
+end:
+  thd->where= save_where;
+  context->table_list= save_list;
+  DBUG_RETURN(result);
+}
 
 /*
   This routine allocates an array for all list constants to achieve a fast
@@ -706,6 +870,7 @@ int partition_info::list_part_cmp(const void* a, const void* b)
 
   SYNOPSIS
     check_list_constants()
+    thd                            Thread object
 
   RETURN VALUE
     TRUE                  An error occurred during creation of list constants
@@ -718,15 +883,18 @@ int partition_info::list_part_cmp(const void* a, const void* b)
     called for LIST PARTITIONed tables.
 */
 
-bool partition_info::check_list_constants()
+bool partition_info::check_list_constants(THD *thd)
 {
-  uint i;
+  uint i, size_entries, no_column_values;
   uint list_index= 0;
   part_elem_value *list_value;
   bool result= TRUE;
-  longlong curr_value, prev_value, type_add, calc_value;
+  longlong type_add, calc_value;
+  void *curr_value, *prev_value;
   partition_element* part_def;
   bool found_null= FALSE;
+  int (*compare_func)(const void *, const void*);
+  void *ptr;
   List_iterator<partition_element> list_func_it(partitions);
   DBUG_ENTER("partition_info::check_list_constants");
 
@@ -767,48 +935,86 @@ bool partition_info::check_list_constants()
       no_list_values++;
   } while (++i < no_parts);
   list_func_it.rewind();
-  list_array= (LIST_PART_ENTRY*)sql_alloc((no_list_values+1) *
-                                          sizeof(LIST_PART_ENTRY));
-  if (unlikely(list_array == NULL))
+  no_column_values= part_field_list.elements;
+  size_entries= column_list ?
+        (no_column_values * sizeof(part_column_list_val)) :
+        sizeof(LIST_PART_ENTRY);
+  ptr= sql_calloc((no_list_values+1) * size_entries);
+  if (unlikely(ptr == NULL))
   {
-    mem_alloc_error(no_list_values * sizeof(LIST_PART_ENTRY));
+    mem_alloc_error(no_list_values * size_entries);
     goto end;
   }
-
-  i= 0;
-  /*
-    Fix to be able to reuse signed sort functions also for unsigned
-    partition functions.
-  */
-  type_add= (longlong)(part_expr->unsigned_flag ?
+  if (column_list)
+  {
+    part_column_list_val *loc_list_col_array;
+    loc_list_col_array= (part_column_list_val*)ptr;
+    list_col_array= (part_column_list_val*)ptr;
+    compare_func= compare_column_values;
+    i= 0;
+    /*
+      Fix to be able to reuse signed sort functions also for unsigned
+      partition functions.
+    */
+    do
+    {
+      part_def= list_func_it++;
+      List_iterator<part_elem_value> list_val_it2(part_def->list_val_list);
+      while ((list_value= list_val_it2++))
+      {
+        part_column_list_val *col_val= list_value->col_val_array;
+        if (unlikely(fix_column_value_functions(thd, col_val, i)))
+        {
+          DBUG_RETURN(TRUE);
+        }
+        memcpy(loc_list_col_array, (const void*)col_val, size_entries);
+        loc_list_col_array+= no_column_values;
+      }
+    } while (++i < no_parts);
+  }
+  else
+  {
+    compare_func= list_part_cmp;
+    list_array= (LIST_PART_ENTRY*)ptr;
+    i= 0;
+    /*
+      Fix to be able to reuse signed sort functions also for unsigned
+      partition functions.
+    */
+    type_add= (longlong)(part_expr->unsigned_flag ?
                                        0x8000000000000000ULL :
                                        0ULL);
 
-  do
-  {
-    part_def= list_func_it++;
-    List_iterator<part_elem_value> list_val_it2(part_def->list_val_list);
-    while ((list_value= list_val_it2++))
+    do
     {
-      calc_value= list_value->value - type_add;
-      list_array[list_index].list_value= calc_value;
-      list_array[list_index++].partition_id= i;
-    }
-  } while (++i < no_parts);
-
+      part_def= list_func_it++;
+      List_iterator<part_elem_value> list_val_it2(part_def->list_val_list);
+      while ((list_value= list_val_it2++))
+      {
+        calc_value= list_value->value - type_add;
+        list_array[list_index].list_value= calc_value;
+        list_array[list_index++].partition_id= i;
+      }
+    } while (++i < no_parts);
+  }
   if (fixed && no_list_values)
   {
     bool first= TRUE;
+    /*
+      list_array and list_col_array are unions, so this works for both
+      variants of LIST partitioning.
+    */
     my_qsort((void*)list_array, no_list_values, sizeof(LIST_PART_ENTRY), 
              &list_part_cmp);
- 
+
     i= 0;
     LINT_INIT(prev_value);
     do
     {
       DBUG_ASSERT(i < no_list_values);
-      curr_value= list_array[i].list_value;
-      if (likely(first || prev_value != curr_value))
+      curr_value= column_list ? (void*)&list_col_array[no_column_values * i] :
+                                (void*)&list_array[i];
+      if (likely(first || compare_func(curr_value, prev_value)))
       {
         prev_value= curr_value;
         first= FALSE;
@@ -831,10 +1037,11 @@ end:
 
   SYNOPSIS
     check_partition_info()
+    thd                 Thread object
+    eng_type            Return value for used engine in partitions
     file                A reference to a handler of the table
     info                Create info
-    engine_type         Return value for used engine in partitions
-    check_partition_function Should we check the partition function
+    add_or_reorg_part   Is it ALTER TABLE ADD/REORGANIZE command
 
   RETURN VALUE
     TRUE                 Error, something went wrong
@@ -850,7 +1057,7 @@ end:
 
 bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
                                           handler *file, HA_CREATE_INFO *info,
-                                          bool check_partition_function)
+                                          bool add_or_reorg_part)
 {
   handlerton *table_engine= default_engine_type;
   uint i, tot_partitions;
@@ -861,11 +1068,11 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
 
   DBUG_PRINT("info", ("default table_engine = %s",
                       ha_resolve_storage_engine_name(table_engine)));
-  if (check_partition_function)
+  if (!add_or_reorg_part)
   {
     int err= 0;
 
-    if (part_type != HASH_PARTITION || !list_of_part_fields)
+    if (!list_of_part_fields)
     {
       DBUG_ASSERT(part_expr);
       err= part_expr->walk(&Item::check_partition_func_processor, 0,
@@ -1064,10 +1271,12 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
     list constants.
   */
 
-  if (fixed)
+  if (add_or_reorg_part)
   {
-    if (unlikely((part_type == RANGE_PARTITION && check_range_constants()) ||
-                  (part_type == LIST_PARTITION && check_list_constants())))
+    if (unlikely((part_type == RANGE_PARTITION &&
+                  check_range_constants(thd)) ||
+                 (part_type == LIST_PARTITION &&
+                  check_list_constants(thd))))
       goto end;
   }
   result= FALSE;
@@ -1098,20 +1307,96 @@ void partition_info::print_no_partition_found(TABLE *table)
 
   if (check_single_table_access(current_thd,
                                 SELECT_ACL, &table_list, TRUE))
+  {
     my_message(ER_NO_PARTITION_FOR_GIVEN_VALUE,
                ER(ER_NO_PARTITION_FOR_GIVEN_VALUE_SILENT), MYF(0));
+  }
   else
   {
-    my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set);
-    if (part_expr->null_value)
-      buf_ptr= (char*)"NULL";
+    if (column_list)
+      buf_ptr= (char*)"from column_list";
     else
-      longlong2str(err_value, buf,
-                   part_expr->unsigned_flag ? 10 : -10);
+    {
+      my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set);
+      if (part_expr->null_value)
+        buf_ptr= (char*)"NULL";
+      else
+        longlong2str(err_value, buf,
+                     part_expr->unsigned_flag ? 10 : -10);
+      dbug_tmp_restore_column_map(table->read_set, old_map);
+    }
     my_error(ER_NO_PARTITION_FOR_GIVEN_VALUE, MYF(0), buf_ptr);
-    dbug_tmp_restore_column_map(table->read_set, old_map);
   }
 }
+
+
+/*
+  Create a new column value in current list
+  SYNOPSIS
+    add_column_value()
+  RETURN
+    >0                 A part_column_list_val object which have been
+                       inserted into its list
+    0                  Memory allocation failure
+*/
+
+part_column_list_val *partition_info::add_column_value()
+{
+  uint max_val= num_columns ? num_columns : MAX_REF_PARTS;
+  DBUG_ENTER("add_column_value");
+  DBUG_PRINT("enter", ("num_columns = %u, curr_list_object %u, max_val = %u",
+                        num_columns, curr_list_object, max_val));
+  if (curr_list_object < max_val)
+  {
+    DBUG_RETURN(&curr_list_val->col_val_array[curr_list_object++]);
+  }
+  my_error(ER_PARTITION_COLUMN_LIST_ERROR, MYF(0));
+  DBUG_RETURN(NULL);
+}
+
+
+/*
+  Set fields related to partition expression
+  SYNOPSIS
+    set_part_expr()
+    start_token               Start of partition function string
+    item_ptr                  Pointer to item tree
+    end_token                 End of partition function string
+    is_subpart                Subpartition indicator
+  RETURN VALUES
+    TRUE                      Memory allocation error
+    FALSE                     Success
+*/
+
+bool partition_info::set_part_expr(char *start_token, Item *item_ptr,
+                                   char *end_token, bool is_subpart)
+{
+  uint expr_len= end_token - start_token;
+  char *func_string= (char*) sql_memdup(start_token, expr_len);
+
+  if (!func_string)
+  {
+    mem_alloc_error(expr_len);
+    return TRUE;
+  }
+  if (is_subpart)
+  {
+    list_of_subpart_fields= FALSE;
+    subpart_expr= item_ptr;
+    subpart_func_string= func_string;
+    subpart_func_len= expr_len;
+  }
+  else
+  {
+    list_of_part_fields= FALSE;
+    part_expr= item_ptr;
+    part_func_string= func_string;
+    part_func_len= expr_len;
+  }
+  return FALSE;
+}
+
+
 /*
   Set up buffers and arrays for fields requiring preparation
   SYNOPSIS
@@ -1223,46 +1508,6 @@ bool partition_info::set_up_charset_field_preps()
     }
     subpart_charset_field_array[i]= NULL;
   }
-  if (tot_fields)
-  {
-    uint k;
-    size= tot_fields*sizeof(char**);
-    if (!(char_ptrs= (uchar**)sql_calloc(size)))
-      goto error;
-    full_part_field_buffers= char_ptrs;
-    if (!(char_ptrs= (uchar**)sql_calloc(size)))
-      goto error;
-    restore_full_part_field_ptrs= char_ptrs;
-    size= (tot_fields + 1) * sizeof(char**);
-    if (!(char_ptrs= (uchar**)sql_calloc(size)))
-      goto error;
-    full_part_charset_field_array= (Field**)char_ptrs;
-    for (i= 0; i < tot_part_fields; i++)
-    {
-      full_part_charset_field_array[i]= part_charset_field_array[i];
-      full_part_field_buffers[i]= part_field_buffers[i];
-    }
-    k= tot_part_fields;
-    for (i= 0; i < tot_subpart_fields; i++)
-    {
-      uint j;
-      bool found= FALSE;
-      field= subpart_charset_field_array[i];
-
-      for (j= 0; j < tot_part_fields; j++)
-      {
-        if (field == part_charset_field_array[i])
-          found= TRUE;
-      }
-      if (!found)
-      {
-        full_part_charset_field_array[k]= subpart_charset_field_array[i];
-        full_part_field_buffers[k]= subpart_field_buffers[i];
-        k++;
-      }
-    }
-    full_part_charset_field_array[k]= NULL;
-  }
   DBUG_RETURN(FALSE);
 error:
   mem_alloc_error(size);
diff --git a/sql/partition_info.h b/sql/partition_info.h
index 415f955d5d4..f232e761946 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -19,6 +19,8 @@
 
 #include "partition_element.h"
 
+#define MAX_STR_SIZE_PF 512
+
 class partition_info;
 
 /* Some function typedefs */
@@ -64,10 +66,9 @@ public:
   /*
     When we have various string fields we might need some preparation
     before and clean-up after calling the get_part_id_func's. We need
-    one such method for get_partition_id and one for
-    get_part_partition_id and one for get_subpartition_id.
+    one such method for get_part_partition_id and one for
+    get_subpartition_id.
   */
-  get_part_id_func get_partition_id_charset;
   get_part_id_func get_part_partition_id_charset;
   get_subpart_id_func get_subpartition_id_charset;
 
@@ -81,7 +82,6 @@ public:
     without duplicates, NULL-terminated.
   */
   Field **full_part_field_array;
-  Field **full_part_charset_field_array;
   /*
     Set of all fields used in partition and subpartition expression.
     Required for testing of partition fields in write_set when
@@ -97,10 +97,8 @@ public:
   */
   uchar **part_field_buffers;
   uchar **subpart_field_buffers;
-  uchar **full_part_field_buffers;
   uchar **restore_part_field_ptrs;
   uchar **restore_subpart_field_ptrs;
-  uchar **restore_full_part_field_ptrs;
 
   Item *part_expr;
   Item *subpart_expr;
@@ -124,6 +122,8 @@ public:
   union {
     longlong *range_int_array;
     LIST_PART_ENTRY *list_array;
+    part_column_list_val *range_col_array;
+    part_column_list_val *list_col_array;
   };
   
   /********************************************
@@ -154,6 +154,10 @@ public:
 
   partition_element *curr_part_elem;
   partition_element *current_partition;
+  part_elem_value *curr_list_val;
+  uint curr_list_object;
+  uint num_columns;
+
   /*
     These key_map's are used for Partitioning to enable quick decisions
     on whether we can derive more information about which partition to
@@ -205,7 +209,7 @@ public:
   bool is_auto_partitioned;
   bool from_openfrm;
   bool has_null_value;
-
+  bool column_list;
 
   partition_info()
   : get_partition_id(NULL), get_part_partition_id(NULL),
@@ -214,11 +218,8 @@ public:
     part_charset_field_array(NULL),
     subpart_charset_field_array(NULL),
     full_part_field_array(NULL),
-    full_part_charset_field_array(NULL),
     part_field_buffers(NULL), subpart_field_buffers(NULL),
-    full_part_field_buffers(NULL),
     restore_part_field_ptrs(NULL), restore_subpart_field_ptrs(NULL),
-    restore_full_part_field_ptrs(NULL),
     part_expr(NULL), subpart_expr(NULL), item_free_list(NULL),
     first_log_entry(NULL), exec_log_entry(NULL), frm_log_entry(NULL),
     list_array(NULL), err_value(0),
@@ -226,6 +227,7 @@ public:
     part_func_string(NULL), subpart_func_string(NULL),
     part_state(NULL),
     curr_part_elem(NULL), current_partition(NULL),
+    curr_list_object(0), num_columns(0),
     default_engine_type(NULL),
     part_result_type(INT_RESULT),
     part_type(NOT_A_PARTITION), subpart_type(NOT_A_PARTITION),
@@ -241,7 +243,7 @@ public:
     list_of_part_fields(FALSE), list_of_subpart_fields(FALSE),
     linear_hash_ind(FALSE), fixed(FALSE),
     is_auto_partitioned(FALSE), from_openfrm(FALSE),
-    has_null_value(FALSE)
+    has_null_value(FALSE), column_list(FALSE)
   {
     all_fields_in_PF.clear_all();
     all_fields_in_PPF.clear_all();
@@ -271,16 +273,19 @@ public:
                                         uint start_no);
   char *has_unique_names();
   bool check_engine_mix(handlerton *engine_type, bool default_engine);
-  bool check_range_constants();
-  bool check_list_constants();
+  bool check_range_constants(THD *thd);
+  bool check_list_constants(THD *thd);
   bool check_partition_info(THD *thd, handlerton **eng_type,
                             handler *file, HA_CREATE_INFO *info,
                             bool check_partition_function);
   void print_no_partition_found(TABLE *table);
+  part_column_list_val *add_column_value();
+  bool set_part_expr(char *start_token, Item *item_ptr,
+                     char *end_token, bool is_subpart);
+  static int compare_column_values(const void *a, const void *b);
   bool set_up_charset_field_preps();
 private:
   static int list_part_cmp(const void* a, const void* b);
-  static int list_part_cmp_unsigned(const void* a, const void* b);
   bool set_up_default_partitions(handler *file, HA_CREATE_INFO *info,
                                  uint start_no);
   bool set_up_default_subpartitions(handler *file, HA_CREATE_INFO *info);
@@ -288,6 +293,9 @@ private:
                                        uint start_no);
   char *create_subpartition_name(uint subpart_no, const char *part_name);
   bool has_unique_name(partition_element *element);
+  bool fix_column_value_functions(THD *thd,
+                                  part_column_list_val *col_val,
+                                  uint part_id);
 };
 
 uint32 get_next_partition_id_range(struct st_partition_iter* part_iter);
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 5531ee71620..514dc06728d 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6180,6 +6180,16 @@ ER_TOO_LONG_FIELD_COMMENT
 
 ER_FUNC_INEXISTENT_NAME_COLLISION 42000 
   eng "FUNCTION %s does not exist. Check the 'Function Name Parsing and Resolution' section in the Reference Manual"
+ER_GLOBAL_PARTITION_INDEX_ERROR
+  eng "Partitioning of indexes only supported for global indexes"
+ER_PARTITION_COLUMN_LIST_ERROR
+  eng "Inconsistency in usage of column lists for partitioning"
+ER_WRONG_TYPE_COLUMN_VALUE_ERROR
+  eng "Partition column values of incorrect type"
+ER_TOO_MANY_PARTITION_FUNC_FIELDS_ERROR
+  eng "Too many fields in '%s'"
+ER_MAXVALUE_IN_LIST_PARTITIONING_ERROR
+  eng "Cannot use MAXVALUE as value in List partitioning"
 
 # When updating these, please update EXPLAIN_FILENAME_MAX_EXTRA_LENGTH in
 # mysql_priv.h with the new maximal additional length for explain_filename.
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 2adbc44eb12..61f243ece1c 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -323,6 +323,7 @@ void lex_start(THD *thd)
   lex->select_lex.select_number= 1;
   lex->length=0;
   lex->part_info= 0;
+  lex->global_flag= 0;
   lex->select_lex.in_sum_expr=0;
   lex->select_lex.ftfunc_list_alloc.empty();
   lex->select_lex.ftfunc_list= &lex->select_lex.ftfunc_list_alloc;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 76fd5354c51..d714e3d0441 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1565,6 +1565,9 @@ typedef struct st_lex : public Query_tables_list
   /* Partition info structure filled in by PARTITION BY parse part */
   partition_info *part_info;
 
+  /* Flag to index a global index created */
+  bool global_flag;
+
   /*
     The definer of the object being created (view, trigger, stored routine).
     I.e. the value of DEFINER clause.
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 61766e5c509..9684e842d40 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -18,16 +18,29 @@
   to partitioning introduced in MySQL version 5.1. It contains functionality
   used by all handlers that support partitioning, such as
   the partitioning handler itself and the NDB handler.
+  (Much of the code in this file has been split into partition_info.cc and
+   the header files partition_info.h + partition_element.h + sql_partition.h)
 
-  The first version was written by Mikael Ronstrom.
+  The first version was written by Mikael Ronstrom 2004-2006.
+  Various parts of the optimizer code was written by Sergey Petrunia.
+  Code have been maintained by Mattias Jonsson.
+  The second version was written by Mikael Ronstrom 2006-2007 with some
+  final fixes for partition pruning in 2008-2009 with assistance from Sergey
+  Petrunia and Mattias Jonsson.
 
-  This version supports RANGE partitioning, LIST partitioning, HASH
+  The first version supports RANGE partitioning, LIST partitioning, HASH
   partitioning and composite partitioning (hereafter called subpartitioning)
   where each RANGE/LIST partitioning is HASH partitioned. The hash function
   can either be supplied by the user or by only a list of fields (also
   called KEY partitioning), where the MySQL server will use an internal
   hash function.
   There are quite a few defaults that can be used as well.
+
+  The second version introduces a new variant of RANGE and LIST partitioning
+  which is often referred to as column lists in the code variables. This
+  enables a user to specify a set of columns and their concatenated value
+  as the partition value. By comparing the concatenation of these values
+  the proper partition can be choosen.
 */
 
 /* Some general useful functions */
@@ -50,9 +63,11 @@ const LEX_STRING partition_keywords[]=
   { C_STRING_WITH_LEN("LIST") }, 
   { C_STRING_WITH_LEN("KEY") },
   { C_STRING_WITH_LEN("MAXVALUE") },
-  { C_STRING_WITH_LEN("LINEAR ") }
+  { C_STRING_WITH_LEN("LINEAR ") },
+  { C_STRING_WITH_LEN(" COLUMN_LIST") }
 };
 static const char *part_str= "PARTITION";
+static const char *subpart_str= "SUBPARTITION";
 static const char *sub_str= "SUB";
 static const char *by_str= "BY";
 static const char *space_str= " ";
@@ -61,26 +76,23 @@ static const char *end_paren_str= ")";
 static const char *begin_paren_str= "(";
 static const char *comma_str= ",";
 
-static int get_part_id_charset_func_all(partition_info *part_info,
-                                        uint32 *part_id,
-                                        longlong *func_value);
-static int get_part_id_charset_func_part(partition_info *part_info,
-                                         uint32 *part_id,
-                                         longlong *func_value);
-static int get_part_id_charset_func_subpart(partition_info *part_info,
-                                            uint32 *part_id,
-                                            longlong *func_value);
-static int get_part_part_id_charset_func(partition_info *part_info,
-                                         uint32 *part_id,
-                                         longlong *func_value);
-static int get_subpart_id_charset_func(partition_info *part_info,
-                                       uint32 *part_id);
+int get_partition_id_list_col(partition_info *part_info,
+                              uint32 *part_id,
+                              longlong *func_value);
 int get_partition_id_list(partition_info *part_info,
                           uint32 *part_id,
                           longlong *func_value);
+int get_partition_id_range_col(partition_info *part_info,
+                               uint32 *part_id,
+                               longlong *func_value);
 int get_partition_id_range(partition_info *part_info,
                            uint32 *part_id,
                            longlong *func_value);
+static int get_part_id_charset_func_part(partition_info *part_info,
+                                         uint32 *part_id,
+                                         longlong *func_value);
+static int get_part_id_charset_func_subpart(partition_info *part_info,
+                                            uint32 *part_id);
 int get_partition_id_hash_nosub(partition_info *part_info,
                                 uint32 *part_id,
                                 longlong *func_value);
@@ -93,30 +105,9 @@ int get_partition_id_linear_hash_nosub(partition_info *part_info,
 int get_partition_id_linear_key_nosub(partition_info *part_info,
                                       uint32 *part_id,
                                       longlong *func_value);
-int get_partition_id_range_sub_hash(partition_info *part_info,
-                                    uint32 *part_id,
-                                    longlong *func_value);
-int get_partition_id_range_sub_key(partition_info *part_info,
-                                   uint32 *part_id,
-                                   longlong *func_value);
-int get_partition_id_range_sub_linear_hash(partition_info *part_info,
-                                           uint32 *part_id,
-                                           longlong *func_value);
-int get_partition_id_range_sub_linear_key(partition_info *part_info,
-                                          uint32 *part_id,
-                                          longlong *func_value);
-int get_partition_id_list_sub_hash(partition_info *part_info,
-                                   uint32 *part_id,
-                                   longlong *func_value);
-int get_partition_id_list_sub_key(partition_info *part_info,
-                                  uint32 *part_id,
-                                  longlong *func_value);
-int get_partition_id_list_sub_linear_hash(partition_info *part_info,
-                                          uint32 *part_id,
-                                          longlong *func_value);
-int get_partition_id_list_sub_linear_key(partition_info *part_info,
-                                         uint32 *part_id,
-                                         longlong *func_value);
+int get_partition_id_with_sub(partition_info *part_info,
+                              uint32 *part_id,
+                              longlong *func_value);
 int get_partition_id_hash_sub(partition_info *part_info,
                               uint32 *part_id); 
 int get_partition_id_key_sub(partition_info *part_info,
@@ -134,14 +125,29 @@ uint32 get_next_partition_id_range(PARTITION_ITERATOR* part_iter);
 uint32 get_next_partition_id_list(PARTITION_ITERATOR* part_iter);
 int get_part_iter_for_interval_via_mapping(partition_info *part_info,
                                            bool is_subpart,
+                                           uint32 *store_length_array,
                                            uchar *min_value, uchar *max_value,
+                                           uint min_len, uint max_len,
                                            uint flags,
                                            PARTITION_ITERATOR *part_iter);
+int get_part_iter_for_interval_cols_via_map(partition_info *part_info,
+                                            bool is_subpart,
+                                            uint32 *store_length_array,
+                                            uchar *min_value, uchar *max_value,
+                                            uint min_len, uint max_len,
+                                            uint flags,
+                                            PARTITION_ITERATOR *part_iter);
 int get_part_iter_for_interval_via_walking(partition_info *part_info,
                                            bool is_subpart,
+                                           uint32 *store_length_array,
                                            uchar *min_value, uchar *max_value,
+                                           uint min_len, uint max_len,
                                            uint flags,
                                            PARTITION_ITERATOR *part_iter);
+static int cmp_rec_and_tuple(part_column_list_val *val, uint32 nvals_in_rec);
+static int cmp_rec_and_tuple_prune(part_column_list_val *val,
+                                   uint32 n_vals_in_rec,
+                                   bool tail_is_min);
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 /*
@@ -161,7 +167,7 @@ bool is_name_in_list(char *name,
                           List<char> list_names)
 {
   List_iterator<char> names_it(list_names);
-  uint no_names= list_names.elements;
+  uint num_names= list_names.elements;
   uint i= 0;
 
   do
@@ -169,7 +175,7 @@ bool is_name_in_list(char *name,
     char *list_name= names_it++;
     if (!(my_strcasecmp(system_charset_info, name, list_name)))
       return TRUE;
-  } while (++i < no_names);
+  } while (++i < num_names);
   return FALSE;
 }
 
@@ -451,6 +457,7 @@ static bool set_up_field_array(TABLE *table,
   uint no_fields= 0;
   uint size_field_array;
   uint i= 0;
+  uint inx;
   partition_info *part_info= table->part_info;
   int result= FALSE;
   DBUG_ENTER("set_up_field_array");
@@ -461,6 +468,16 @@ static bool set_up_field_array(TABLE *table,
     if (field->flags & GET_FIXED_FIELDS_FLAG)
       no_fields++;
   }
+  if (no_fields > MAX_REF_PARTS)
+  {
+    char *ptr;
+    if (is_sub_part)
+      ptr= (char*)"subpartition function";
+    else
+      ptr= (char*)"partition function";
+    my_error(ER_TOO_MANY_PARTITION_FUNC_FIELDS_ERROR, MYF(0), ptr);
+    DBUG_RETURN(TRUE);
+  }
   if (no_fields == 0)
   {
     /*
@@ -470,7 +487,7 @@ static bool set_up_field_array(TABLE *table,
     DBUG_RETURN(result);
   }
   size_field_array= (no_fields+1)*sizeof(Field*);
-  field_array= (Field**)sql_alloc(size_field_array);
+  field_array= (Field**)sql_calloc(size_field_array);
   if (unlikely(!field_array))
   {
     mem_alloc_error(size_field_array);
@@ -485,7 +502,30 @@ static bool set_up_field_array(TABLE *table,
       field->flags|= FIELD_IN_PART_FUNC_FLAG;
       if (likely(!result))
       {
-        field_array[i++]= field;
+        if (!is_sub_part && part_info->column_list)
+        {
+          List_iterator<char> it(part_info->part_field_list);
+          char *field_name;
+
+          DBUG_ASSERT(no_fields == part_info->part_field_list.elements);
+          inx= 0;
+          do
+          {
+            field_name= it++;
+            if (!strcmp(field_name, field->field_name))
+              break;
+          } while (++inx < no_fields);
+          if (inx == no_fields)
+          {
+            mem_alloc_error(1);
+            result= TRUE;
+            continue;
+          }
+        }
+        else
+          inx= i;
+        field_array[inx]= field;
+        i++;
 
         /*
           We check that the fields are proper. It is required for each
@@ -564,7 +604,7 @@ static bool create_full_part_field_array(THD *thd, TABLE *table,
         no_part_fields++;
     }
     size_field_array= (no_part_fields+1)*sizeof(Field*);
-    field_array= (Field**)sql_alloc(size_field_array);
+    field_array= (Field**)sql_calloc(size_field_array);
     if (unlikely(!field_array))
     {
       mem_alloc_error(size_field_array);
@@ -776,7 +816,7 @@ static bool handle_list_of_fields(List_iterator<char> it,
       goto end;
     }
   }
-  if (is_list_empty)
+  if (is_list_empty && part_info->part_type == HASH_PARTITION)
   {
     uint primary_key= table->s->primary_key;
     if (primary_key != MAX_KEY)
@@ -868,7 +908,6 @@ int check_signed_flag(partition_info *part_info)
     table                The table object
     part_info            Reference to partitioning data structure
     is_sub_part          Is the table subpartitioned as well
-    is_field_to_be_setup Flag if we are to set-up field arrays
 
   RETURN VALUE
     TRUE                 An error occurred, something was wrong with the
@@ -891,8 +930,8 @@ int check_signed_flag(partition_info *part_info)
     on the field object.
 */
 
-bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
-                          bool is_sub_part, bool is_field_to_be_setup)
+static bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
+                                 bool is_sub_part)
 {
   partition_info *part_info= table->part_info;
   uint dir_length, home_dir_length;
@@ -985,8 +1024,7 @@ bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   if (unlikely(error))
   {
     DBUG_PRINT("info", ("Field in partition function not part of table"));
-    if (is_field_to_be_setup)
-      clear_field_flag(table);
+    clear_field_flag(table);
     goto end;
   }
   thd->where= save_where;
@@ -998,9 +1036,7 @@ bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   }
   if ((!is_sub_part) && (error= check_signed_flag(part_info)))
     goto end;
-  result= FALSE;
-  if (is_field_to_be_setup)
-    result= set_up_field_array(table, is_sub_part);
+  result= set_up_field_array(table, is_sub_part);
   if (!is_sub_part)
     part_info->fixed= TRUE;
 end:
@@ -1283,64 +1319,47 @@ static void set_up_partition_func_pointers(partition_info *part_info)
 
   if (part_info->is_sub_partitioned())
   {
+    part_info->get_partition_id= get_partition_id_with_sub;
     if (part_info->part_type == RANGE_PARTITION)
     {
-      part_info->get_part_partition_id= get_partition_id_range;
+      if (part_info->column_list)
+        part_info->get_part_partition_id= get_partition_id_range_col;
+      else
+        part_info->get_part_partition_id= get_partition_id_range;
       if (part_info->list_of_subpart_fields)
       {
         if (part_info->linear_hash_ind)
-        {
-          part_info->get_partition_id= get_partition_id_range_sub_linear_key;
           part_info->get_subpartition_id= get_partition_id_linear_key_sub;
-        }
         else
-        {
-          part_info->get_partition_id= get_partition_id_range_sub_key;
           part_info->get_subpartition_id= get_partition_id_key_sub;
-        }
       }
       else
       {
         if (part_info->linear_hash_ind)
-        {
-          part_info->get_partition_id= get_partition_id_range_sub_linear_hash;
           part_info->get_subpartition_id= get_partition_id_linear_hash_sub;
-        }
         else
-        {
-          part_info->get_partition_id= get_partition_id_range_sub_hash;
           part_info->get_subpartition_id= get_partition_id_hash_sub;
-        }
       }
     }
     else /* LIST Partitioning */
     {
-      part_info->get_part_partition_id= get_partition_id_list;
+      if (part_info->column_list)
+        part_info->get_part_partition_id= get_partition_id_list_col;
+      else
+        part_info->get_part_partition_id= get_partition_id_list;
       if (part_info->list_of_subpart_fields)
       {
         if (part_info->linear_hash_ind)
-        {
-          part_info->get_partition_id= get_partition_id_list_sub_linear_key;
           part_info->get_subpartition_id= get_partition_id_linear_key_sub;
-        }
         else
-        {
-          part_info->get_partition_id= get_partition_id_list_sub_key;
           part_info->get_subpartition_id= get_partition_id_key_sub;
-        }
       }
       else
       {
         if (part_info->linear_hash_ind)
-        {
-          part_info->get_partition_id= get_partition_id_list_sub_linear_hash;
           part_info->get_subpartition_id= get_partition_id_linear_hash_sub;
-        }
         else
-        {
-          part_info->get_partition_id= get_partition_id_list_sub_hash;
           part_info->get_subpartition_id= get_partition_id_hash_sub;
-        }
       }
     }
   }
@@ -1349,9 +1368,19 @@ static void set_up_partition_func_pointers(partition_info *part_info)
     part_info->get_part_partition_id= NULL;
     part_info->get_subpartition_id= NULL;
     if (part_info->part_type == RANGE_PARTITION)
-      part_info->get_partition_id= get_partition_id_range;
+    {
+      if (part_info->column_list)
+        part_info->get_partition_id= get_partition_id_range_col;
+      else
+        part_info->get_partition_id= get_partition_id_range;
+    }
     else if (part_info->part_type == LIST_PARTITION)
-      part_info->get_partition_id= get_partition_id_list;
+    {
+      if (part_info->column_list)
+        part_info->get_partition_id= get_partition_id_list_col;
+      else
+        part_info->get_partition_id= get_partition_id_list;
+    }
     else /* HASH partitioning */
     {
       if (part_info->list_of_part_fields)
@@ -1370,32 +1399,37 @@ static void set_up_partition_func_pointers(partition_info *part_info)
       }
     }
   }
-  if (part_info->full_part_charset_field_array)
+  /*
+    We need special functions to handle character sets since they require copy
+    of field pointers and restore afterwards. For subpartitioned tables we do
+    the copy and restore individually on the part and subpart parts. For non-
+    subpartitioned tables we use the same functions as used for the parts part
+    of subpartioning.
+    Thus for subpartitioned tables the get_partition_id is always
+    get_partition_id_with_sub, even when character sets exists.
+  */
+  if (part_info->part_charset_field_array)
   {
-    DBUG_ASSERT(part_info->get_partition_id);
-    part_info->get_partition_id_charset= part_info->get_partition_id;
-    if (part_info->part_charset_field_array &&
-        part_info->subpart_charset_field_array)
-      part_info->get_partition_id= get_part_id_charset_func_all;
-    else if (part_info->part_charset_field_array)
-      part_info->get_partition_id= get_part_id_charset_func_part;
+    if (part_info->is_sub_partitioned())
+    {
+      DBUG_ASSERT(part_info->get_part_partition_id);
+      part_info->get_part_partition_id_charset=
+           part_info->get_part_partition_id;
+      part_info->get_part_partition_id= get_part_id_charset_func_part;
+    }
     else
-      part_info->get_partition_id= get_part_id_charset_func_subpart;
-  }
-  if (part_info->part_charset_field_array &&
-      part_info->is_sub_partitioned())
-  {
-    DBUG_ASSERT(part_info->get_part_partition_id);
-    part_info->get_part_partition_id_charset=
-          part_info->get_part_partition_id;
-    part_info->get_part_partition_id= get_part_part_id_charset_func;
+    {
+      DBUG_ASSERT(part_info->get_partition_id);
+      part_info->get_part_partition_id_charset= part_info->get_partition_id;
+      part_info->get_partition_id= get_part_id_charset_func_part;
+    }
   }
   if (part_info->subpart_charset_field_array)
   {
     DBUG_ASSERT(part_info->get_subpartition_id);
     part_info->get_subpartition_id_charset=
           part_info->get_subpartition_id;
-    part_info->get_subpartition_id= get_subpart_id_charset_func;
+    part_info->get_subpartition_id= get_part_id_charset_func_subpart;
   }
   DBUG_VOID_RETURN;
 }
@@ -1603,12 +1637,12 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->subpart_expr,
-                                        table, TRUE, TRUE)))
+                                        table, TRUE)))
         goto end;
       if (unlikely(part_info->subpart_expr->result_type() != INT_RESULT))
       {
         my_error(ER_PARTITION_FUNC_NOT_ALLOWED_ERROR, MYF(0),
-                 "SUBPARTITION");
+                 subpart_str);
         goto end;
       }
     }
@@ -1631,7 +1665,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->part_expr,
-                                        table, FALSE, TRUE)))
+                                        table, FALSE)))
         goto end;
       if (unlikely(part_info->part_expr->result_type() != INT_RESULT))
       {
@@ -1644,19 +1678,28 @@ bool fix_partition_func(THD *thd, TABLE *table,
   else
   {
     const char *error_str;
-    if (unlikely(fix_fields_part_func(thd, part_info->part_expr,
-                                      table, FALSE, TRUE)))
-      goto end;
+    if (part_info->column_list)
+    {
+      List_iterator<char> it(part_info->part_field_list);
+      if (unlikely(handle_list_of_fields(it, table, part_info, FALSE)))
+        goto end;
+    }
+    else
+    {
+      if (unlikely(fix_fields_part_func(thd, part_info->part_expr,
+                                        table, FALSE)))
+        goto end;
+    }
     if (part_info->part_type == RANGE_PARTITION)
     {
       error_str= partition_keywords[PKW_RANGE].str; 
-      if (unlikely(part_info->check_range_constants()))
+      if (unlikely(part_info->check_range_constants(thd)))
         goto end;
     }
     else if (part_info->part_type == LIST_PARTITION)
     {
       error_str= partition_keywords[PKW_LIST].str; 
-      if (unlikely(part_info->check_list_constants()))
+      if (unlikely(part_info->check_list_constants(thd)))
         goto end;
     }
     else
@@ -1670,7 +1713,8 @@ bool fix_partition_func(THD *thd, TABLE *table,
       my_error(ER_PARTITIONS_MUST_BE_DEFINED_ERROR, MYF(0), error_str);
       goto end;
     }
-    if (unlikely(part_info->part_expr->result_type() != INT_RESULT))
+    if (unlikely(!part_info->column_list &&
+                  part_info->part_expr->result_type() != INT_RESULT))
     {
       my_error(ER_PARTITION_FUNC_NOT_ALLOWED_ERROR, MYF(0), part_str);
       goto end;
@@ -1723,9 +1767,9 @@ end:
 
 static int add_write(File fptr, const char *buf, uint len)
 {
-  uint len_written= my_write(fptr, (const uchar*)buf, len, MYF(0));
+  uint ret_code= my_write(fptr, (const uchar*)buf, len, MYF(MY_FNABP));
 
-  if (likely(len == len_written))
+  if (likely(ret_code == 0))
     return 0;
   else
     return 1;
@@ -1774,14 +1818,8 @@ static int add_begin_parenthesis(File fptr)
 static int add_part_key_word(File fptr, const char *key_string)
 {
   int err= add_string(fptr, key_string);
-
   err+= add_space(fptr);
-  return err + add_begin_parenthesis(fptr);
-}
-
-static int add_hash(File fptr)
-{
-  return add_part_key_word(fptr, partition_keywords[PKW_HASH].str);
+  return err;
 }
 
 static int add_partition(File fptr)
@@ -1812,15 +1850,15 @@ static int add_subpartition_by(File fptr)
   return err + add_partition_by(fptr);
 }
 
-static int add_key_partition(File fptr, List<char> field_list)
+static int add_part_field_list(File fptr, List<char> field_list)
 {
   uint i, no_fields;
-  int err;
+  int err= 0;
 
   List_iterator<char> part_it(field_list);
-  err= add_part_key_word(fptr, partition_keywords[PKW_KEY].str);
   no_fields= field_list.elements;
   i= 0;
+  err+= add_begin_parenthesis(fptr);
   while (i < no_fields)
   {
     const char *field_str= part_it++;
@@ -1836,6 +1874,7 @@ static int add_key_partition(File fptr, List<char> field_list)
       err+= add_comma(fptr);
     i++;
   }
+  err+= add_end_parenthesis(fptr);
   return err;
 }
 
@@ -1945,24 +1984,88 @@ static int add_partition_options(File fptr, partition_element *p_elem)
   return err + add_engine(fptr,p_elem->engine_type);
 }
 
-static int add_partition_values(File fptr, partition_info *part_info, partition_element *p_elem)
+static int add_column_list_values(File fptr, partition_info *part_info,
+                                  part_elem_value *list_value)
+{
+  int err= 0;
+  uint i;
+  uint no_elements= part_info->part_field_list.elements;
+  err+= add_string(fptr, partition_keywords[PKW_COLUMNS].str);
+  err+= add_begin_parenthesis(fptr);
+  for (i= 0; i < no_elements; i++)
+  {
+    part_column_list_val *col_val= &list_value->col_val_array[i];
+    if (col_val->max_value)
+      err+= add_string(fptr, partition_keywords[PKW_MAXVALUE].str);
+    else if (col_val->null_value)
+      err+= add_string(fptr, "NULL");
+    else
+    {
+      char buffer[MAX_STR_SIZE_PF];
+      String str(buffer, sizeof(buffer), &my_charset_bin);
+      Item *item_expr= col_val->item_expression;
+      if (!col_val->fixed &&
+          (item_expr->fix_fields(current_thd, (Item**)0) || 
+           (!item_expr->const_item())))
+      {
+        my_error(ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR, MYF(0));
+        return 1;
+      }
+      col_val->fixed= 1;
+      if (item_expr->null_value)
+        err+= add_string(fptr, "NULL");
+      else
+      {
+        String *res= item_expr->val_str(&str);
+        if (!res)
+        {
+          my_error(ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR, MYF(0));
+          return 1;
+        }
+        if (item_expr->result_type() == STRING_RESULT)
+          err+= add_string(fptr,"'");
+        err+= add_string_object(fptr, res);
+        if (item_expr->result_type() == STRING_RESULT)
+          err+= add_string(fptr,"'");
+      }
+    }
+    if (i != (no_elements - 1))
+      err+= add_string(fptr, comma_str);
+  }
+  err+= add_end_parenthesis(fptr);
+  return err;
+}
+
+static int add_partition_values(File fptr, partition_info *part_info,
+                                partition_element *p_elem)
 {
   int err= 0;
 
   if (part_info->part_type == RANGE_PARTITION)
   {
     err+= add_string(fptr, " VALUES LESS THAN ");
-    if (!p_elem->max_value)
+    if (part_info->column_list)
     {
+      List_iterator<part_elem_value> list_val_it(p_elem->list_val_list);
+      part_elem_value *list_value= list_val_it++;
       err+= add_begin_parenthesis(fptr);
-      if (p_elem->signed_flag)
-        err+= add_int(fptr, p_elem->range_value);
-      else
-        err+= add_uint(fptr, p_elem->range_value);
+      err+= add_column_list_values(fptr, part_info, list_value);
       err+= add_end_parenthesis(fptr);
     }
     else
-      err+= add_string(fptr, partition_keywords[PKW_MAXVALUE].str);
+    {
+      if (!p_elem->max_value)
+      {
+        err+= add_begin_parenthesis(fptr);
+        if (p_elem->signed_flag)
+          err+= add_int(fptr, p_elem->range_value);
+        else
+          err+= add_uint(fptr, p_elem->range_value);
+        err+= add_end_parenthesis(fptr);
+      }
+      else
+        err+= add_string(fptr, partition_keywords[PKW_MAXVALUE].str);
+    }
   }
   else if (part_info->part_type == LIST_PARTITION)
   {
@@ -1987,10 +2090,15 @@ static int add_partition_values(File fptr, partition_info *part_info, partition_
     {
       part_elem_value *list_value= list_val_it++;
 
-      if (!list_value->unsigned_flag)
-        err+= add_int(fptr, list_value->value);
+      if (part_info->column_list)
+        err+= add_column_list_values(fptr, part_info, list_value);
       else
-        err+= add_uint(fptr, list_value->value);
+      {
+        if (!list_value->unsigned_flag)
+          err+= add_int(fptr, list_value->value);
+        else
+          err+= add_uint(fptr, list_value->value);
+      }
       if (i != (no_items-1))
         err+= add_comma(fptr);
     } while (++i < no_items);
@@ -2073,9 +2181,12 @@ char *generate_partition_syntax(partition_info *part_info,
       if (part_info->linear_hash_ind)
         err+= add_string(fptr, partition_keywords[PKW_LINEAR].str);
       if (part_info->list_of_part_fields)
-        err+= add_key_partition(fptr, part_info->part_field_list);
+      {
+        err+= add_part_key_word(fptr, partition_keywords[PKW_KEY].str);
+        err+= add_part_field_list(fptr, part_info->part_field_list);
+      }
       else
-        err+= add_hash(fptr);
+        err+= add_part_key_word(fptr, partition_keywords[PKW_HASH].str);
       break;
     default:
       DBUG_ASSERT(0);
@@ -2085,9 +2196,17 @@ char *generate_partition_syntax(partition_info *part_info,
       DBUG_RETURN(NULL);
   }
   if (part_info->part_expr)
+  {
+    err+= add_begin_parenthesis(fptr);
     err+= add_string_len(fptr, part_info->part_func_string,
                          part_info->part_func_len);
-  err+= add_end_parenthesis(fptr);
+    err+= add_end_parenthesis(fptr);
+  }
+  else if (part_info->column_list)
+  {
+    err+= add_string(fptr, partition_keywords[PKW_COLUMNS].str);
+    err+= add_part_field_list(fptr, part_info->part_field_list);
+  }
   if ((!part_info->use_default_no_partitions) &&
        part_info->use_default_partitions)
   {
@@ -2103,13 +2222,19 @@ char *generate_partition_syntax(partition_info *part_info,
     if (part_info->linear_hash_ind)
       err+= add_string(fptr, partition_keywords[PKW_LINEAR].str);
     if (part_info->list_of_subpart_fields)
-      err+= add_key_partition(fptr, part_info->subpart_field_list);
+    {
+      add_part_key_word(fptr, partition_keywords[PKW_KEY].str);
+      add_part_field_list(fptr, part_info->subpart_field_list);
+    }
     else
-      err+= add_hash(fptr);
+      err+= add_part_key_word(fptr, partition_keywords[PKW_HASH].str);
     if (part_info->subpart_expr)
+    {
+      err+= add_begin_parenthesis(fptr);
       err+= add_string_len(fptr, part_info->subpart_func_string,
                            part_info->subpart_func_len);
-    err+= add_end_parenthesis(fptr);
+      err+= add_end_parenthesis(fptr);
+    }
     if ((!part_info->use_default_no_subpartitions) && 
           part_info->use_default_subpartitions)
     {
@@ -2446,7 +2571,7 @@ static uint32 get_part_id_linear_key(partition_info *part_info,
                                      uint no_parts,
                                      longlong *func_value)
 {
-  DBUG_ENTER("get_partition_id_linear_key");
+  DBUG_ENTER("get_part_id_linear_key");
 
   *func_value= calculate_key_value(field_array);
   DBUG_RETURN(get_part_id_from_linear_hash(*func_value,
@@ -2537,6 +2662,44 @@ static void restore_part_field_pointers(Field **ptr, uchar **restore_ptr)
   return;
 }
 
+/*
+  This function is used to calculate the partition id where all partition
+  fields have been prepared to point to a record where the partition field
+  values are bound.
+
+  SYNOPSIS
+    get_partition_id()
+    part_info           A reference to the partition_info struct where all the
+                        desired information is given
+    out:part_id         The partition id is returned through this pointer
+    out: func_value     Value of partition function (longlong)
+
+  RETURN VALUE
+    part_id                     Partition id of partition that would contain
+                                row with given values of PF-fields
+    HA_ERR_NO_PARTITION_FOUND   The fields of the partition function didn't
+                                fit into any partition and thus the values of 
+                                the PF-fields are not allowed.
+
+  DESCRIPTION
+    A routine used from write_row, update_row and delete_row from any
+    handler supporting partitioning. It is also a support routine for
+    get_partition_set used to find the set of partitions needed to scan
+    for a certain index scan or full table scan.
+    
+    It is actually 9 different variants of this function which are called
+    through a function pointer.
+
+    get_partition_id_list
+    get_partition_id_list_col
+    get_partition_id_range
+    get_partition_id_range_col
+    get_partition_id_hash_nosub
+    get_partition_id_key_nosub
+    get_partition_id_linear_hash_nosub
+    get_partition_id_linear_key_nosub
+    get_partition_id_with_sub
+*/
 
 /*
   This function is used to calculate the main partition to use in the case of
@@ -2558,67 +2721,26 @@ static void restore_part_field_pointers(Field **ptr, uchar **restore_ptr)
 
   DESCRIPTION
     
-    It is actually 6 different variants of this function which are called
+    It is actually 8 different variants of this function which are called
     through a function pointer.
 
     get_partition_id_list
+    get_partition_id_list_col
     get_partition_id_range
+    get_partition_id_range_col
     get_partition_id_hash_nosub
     get_partition_id_key_nosub
-    get_partition_id_linear_hash_nosub
+    get_partition_id_linhash_nosub
     get_partition_id_linear_key_nosub
 */
 
-static int get_part_id_charset_func_subpart(partition_info *part_info,
-                                            uint32 *part_id,
-                                            longlong *func_value)
-{
-  int res;
-  copy_to_part_field_buffers(part_info->subpart_charset_field_array,
-                             part_info->subpart_field_buffers,
-                             part_info->restore_subpart_field_ptrs);
-  res= part_info->get_partition_id_charset(part_info, part_id, func_value);
-  restore_part_field_pointers(part_info->subpart_charset_field_array,
-                              part_info->restore_subpart_field_ptrs);
-  return res;
-}
-
-
 static int get_part_id_charset_func_part(partition_info *part_info,
                                          uint32 *part_id,
                                          longlong *func_value)
 {
   int res;
-  copy_to_part_field_buffers(part_info->part_charset_field_array,
-                             part_info->part_field_buffers,
-                             part_info->restore_part_field_ptrs);
-  res= part_info->get_partition_id_charset(part_info, part_id, func_value);
-  restore_part_field_pointers(part_info->part_charset_field_array,
-                              part_info->restore_part_field_ptrs);
-  return res;
-}
-
-
-static int get_part_id_charset_func_all(partition_info *part_info,
-                                        uint32 *part_id,
-                                        longlong *func_value)
-{
-  int res;
-  copy_to_part_field_buffers(part_info->full_part_field_array,
-                             part_info->full_part_field_buffers,
-                             part_info->restore_full_part_field_ptrs);
-  res= part_info->get_partition_id_charset(part_info, part_id, func_value);
-  restore_part_field_pointers(part_info->full_part_field_array,
-                              part_info->restore_full_part_field_ptrs);
-  return res;
-}
+  DBUG_ENTER("get_part_id_charset_func_part");
 
-
-static int get_part_part_id_charset_func(partition_info *part_info,
-                                         uint32 *part_id,
-                                         longlong *func_value)
-{
-  int res;
   copy_to_part_field_buffers(part_info->part_charset_field_array,
                              part_info->part_field_buffers,
                              part_info->restore_part_field_ptrs);
@@ -2626,21 +2748,58 @@ static int get_part_part_id_charset_func(partition_info *part_info,
                                                 part_id, func_value);
   restore_part_field_pointers(part_info->part_charset_field_array,
                               part_info->restore_part_field_ptrs);
-  return res;
+  DBUG_RETURN(res);
 }
 
 
-static int get_subpart_id_charset_func(partition_info *part_info,
-                                       uint32 *part_id)
+static int get_part_id_charset_func_subpart(partition_info *part_info,
+                                            uint32 *part_id)
 {
   int res;
+  DBUG_ENTER("get_part_id_charset_func_subpart");
+
   copy_to_part_field_buffers(part_info->subpart_charset_field_array,
                              part_info->subpart_field_buffers,
                              part_info->restore_subpart_field_ptrs);
   res= part_info->get_subpartition_id_charset(part_info, part_id);
   restore_part_field_pointers(part_info->subpart_charset_field_array,
                               part_info->restore_subpart_field_ptrs);
-  return res;
+  DBUG_RETURN(res);
+}
+
+int get_partition_id_list_col(partition_info *part_info,
+                              uint32 *part_id,
+                              longlong *func_value)
+{
+  part_column_list_val *list_col_array= part_info->list_col_array;
+  uint no_columns= part_info->part_field_list.elements;
+  int list_index, cmp;
+  int min_list_index= 0;
+  int max_list_index= part_info->no_list_values - 1;
+  DBUG_ENTER("get_partition_id_list_col");
+
+  while (max_list_index >= min_list_index)
+  {
+    list_index= (max_list_index + min_list_index) >> 1;
+    cmp= cmp_rec_and_tuple(list_col_array + list_index*no_columns,
+                          no_columns);
+    if (cmp > 0)
+      min_list_index= list_index + 1;
+    else if (cmp < 0)
+    {
+      if (!list_index)
+        goto notfound;
+      max_list_index= list_index - 1;
+    }
+    else
+    {
+      *part_id= (uint32)list_col_array[list_index].partition_id;
+      DBUG_RETURN(0);
+    }
+  }
+notfound:
+  *part_id= 0;
+  DBUG_RETURN(HA_ERR_NO_PARTITION_FOUND);
 }
 
 
@@ -2735,6 +2894,44 @@ notfound:
     The edge of corresponding sub-array of part_info->list_array
 */
 
+uint32 get_partition_id_cols_list_for_endpoint(partition_info *part_info,
+                                               bool left_endpoint,
+                                               bool include_endpoint,
+                                               uint32 nparts)
+{
+  part_column_list_val *list_col_array= part_info->list_col_array;
+  uint no_columns= part_info->part_field_list.elements;
+  int list_index, cmp;
+  uint min_list_index= 0;
+  uint max_list_index= part_info->no_list_values - 1;
+  bool tailf= !(left_endpoint ^ include_endpoint);
+  DBUG_ENTER("get_partition_id_cols_list_for_endpoint");
+
+  do
+  {
+    list_index= (max_list_index + min_list_index) >> 1;
+    cmp= cmp_rec_and_tuple_prune(list_col_array + list_index*no_columns,
+                                 nparts, tailf);
+    if (cmp > 0)
+      min_list_index= list_index + 1;
+    else if (cmp < 0)
+    {
+      if (!list_index)
+        goto notfound;
+      max_list_index= list_index - 1;
+    }
+    else 
+    {
+      DBUG_RETURN(list_index + test(left_endpoint ^ include_endpoint));
+    }
+  } while (max_list_index >= min_list_index);
+notfound:
+  if (cmp > 0)
+    list_index++;
+  DBUG_RETURN(list_index);
+}
+
+
 uint32 get_list_array_idx_for_endpoint_charset(partition_info *part_info,
                                                bool left_endpoint,
                                                bool include_endpoint)
@@ -2795,6 +2992,44 @@ notfound:
 }
 
 
+int get_partition_id_range_col(partition_info *part_info,
+                               uint32 *part_id,
+                               longlong *func_value)
+{
+  part_column_list_val *range_col_array= part_info->range_col_array;
+  uint no_columns= part_info->part_field_list.elements;
+  uint max_partition= part_info->no_parts - 1;
+  uint min_part_id= 0;
+  uint max_part_id= max_partition;
+  uint loc_part_id;
+  DBUG_ENTER("get_partition_id_range_col");
+
+  while (max_part_id > min_part_id)
+  {
+    loc_part_id= (max_part_id + min_part_id + 1) >> 1;
+    if (cmp_rec_and_tuple(range_col_array + loc_part_id*no_columns,
+                         no_columns) >= 0)
+      min_part_id= loc_part_id + 1;
+    else
+      max_part_id= loc_part_id - 1;
+  }
+  loc_part_id= max_part_id;
+  if (loc_part_id != max_partition)
+    if (cmp_rec_and_tuple(range_col_array + loc_part_id*no_columns,
+                         no_columns) >= 0)
+      loc_part_id++;
+  *part_id= (uint32)loc_part_id;
+  if (loc_part_id == max_partition &&
+      (cmp_rec_and_tuple(range_col_array + loc_part_id*no_columns,
+                        no_columns) >= 0))
+    DBUG_RETURN(HA_ERR_NO_PARTITION_FOUND);
+
+  DBUG_PRINT("exit",("partition: %d", *part_id));
+  DBUG_RETURN(0);
+  return 0;
+}
+
+
 int get_partition_id_range(partition_info *part_info,
                            uint32 *part_id,
                            longlong *func_value)
@@ -2995,8 +3230,8 @@ int get_partition_id_key_nosub(partition_info *part_info,
 
 
 int get_partition_id_linear_key_nosub(partition_info *part_info,
-                                       uint32 *part_id,
-                                       longlong *func_value)
+                                      uint32 *part_id,
+                                      longlong *func_value)
 {
   *part_id= get_part_id_linear_key(part_info,
                                    part_info->part_field_array,
@@ -3005,215 +3240,27 @@ int get_partition_id_linear_key_nosub(partition_info *part_info,
 }
 
 
-int get_partition_id_range_sub_hash(partition_info *part_info,
-                                     uint32 *part_id,
-                                     longlong *func_value)
-{
-  uint32 loc_part_id, sub_part_id;
-  uint no_subparts;
-  longlong local_func_value;
-  int error;
-  DBUG_ENTER("get_partition_id_range_sub_hash");
-  LINT_INIT(loc_part_id);
-  LINT_INIT(sub_part_id);
-
-  if (unlikely((error= get_partition_id_range(part_info, &loc_part_id,
-                                              func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-  no_subparts= part_info->no_subparts;
-  if (unlikely((error= get_part_id_hash(no_subparts, part_info->subpart_expr,
-                                        &sub_part_id, &local_func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-
-  *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
-  DBUG_RETURN(0);
-}
-
-
-int get_partition_id_range_sub_linear_hash(partition_info *part_info,
-                                            uint32 *part_id,
-                                            longlong *func_value)
-{
-  uint32 loc_part_id, sub_part_id;
-  uint no_subparts;
-  longlong local_func_value;
-  int error;
-  DBUG_ENTER("get_partition_id_range_sub_linear_hash");
-  LINT_INIT(loc_part_id);
-  LINT_INIT(sub_part_id);
-
-  if (unlikely((error= get_partition_id_range(part_info, &loc_part_id,
-                                              func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-  no_subparts= part_info->no_subparts;
-  if (unlikely((error= get_part_id_linear_hash(part_info, no_subparts,
-                                               part_info->subpart_expr,
-                                               &sub_part_id,
-                                               &local_func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-
-  *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
-  DBUG_RETURN(0);
-}
-
-
-int get_partition_id_range_sub_key(partition_info *part_info,
-                                    uint32 *part_id,
-                                    longlong *func_value)
-{
-  uint32 loc_part_id, sub_part_id;
-  uint no_subparts;
-  longlong local_func_value;
-  int error;
-  DBUG_ENTER("get_partition_id_range_sub_key");
-  LINT_INIT(loc_part_id);
-
-  if (unlikely((error= get_partition_id_range(part_info, &loc_part_id,
-                                              func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-  no_subparts= part_info->no_subparts;
-  sub_part_id= get_part_id_key(part_info->subpart_field_array,
-                               no_subparts, &local_func_value);
-  *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
-  DBUG_RETURN(0);
-}
-
-
-int get_partition_id_range_sub_linear_key(partition_info *part_info,
-                                           uint32 *part_id,
-                                           longlong *func_value)
-{
-  uint32 loc_part_id, sub_part_id;
-  uint no_subparts;
-  longlong local_func_value;
-  int error;
-  DBUG_ENTER("get_partition_id_range_sub_linear_key");
-  LINT_INIT(loc_part_id);
-
-  if (unlikely((error= get_partition_id_range(part_info, &loc_part_id,
-                                              func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-  no_subparts= part_info->no_subparts;
-  sub_part_id= get_part_id_linear_key(part_info,
-                                      part_info->subpart_field_array,
-                                      no_subparts, &local_func_value);
-  *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
-  DBUG_RETURN(0);
-}
-
-
-int get_partition_id_list_sub_hash(partition_info *part_info,
-                                    uint32 *part_id,
-                                    longlong *func_value)
-{
-  uint32 loc_part_id, sub_part_id;
-  uint no_subparts;
-  longlong local_func_value;
-  int error;
-  DBUG_ENTER("get_partition_id_list_sub_hash");
-  LINT_INIT(sub_part_id);
-
-  if (unlikely((error= get_partition_id_list(part_info, &loc_part_id,
-                                             func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-  no_subparts= part_info->no_subparts;
-  if (unlikely((error= get_part_id_hash(no_subparts, part_info->subpart_expr,
-                                        &sub_part_id, &local_func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-
-  *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
-  DBUG_RETURN(0);
-}
-
-
-int get_partition_id_list_sub_linear_hash(partition_info *part_info,
-                                           uint32 *part_id,
-                                           longlong *func_value)
-{
-  uint32 loc_part_id, sub_part_id;
-  uint no_subparts;
-  longlong local_func_value;
-  int error;
-  DBUG_ENTER("get_partition_id_list_sub_linear_hash");
-  LINT_INIT(sub_part_id);
-
-  if (unlikely((error= get_partition_id_list(part_info, &loc_part_id,
-                                             func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-  no_subparts= part_info->no_subparts;
-  if (unlikely((error= get_part_id_linear_hash(part_info, no_subparts,
-                                               part_info->subpart_expr,
-                                               &sub_part_id,
-                                               &local_func_value))))
-  {
-    DBUG_RETURN(error);
-  }
-   
-  *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
-  DBUG_RETURN(0);
-}
-
-
-int get_partition_id_list_sub_key(partition_info *part_info,
-                                   uint32 *part_id,
-                                   longlong *func_value)
+int get_partition_id_with_sub(partition_info *part_info,
+                              uint32 *part_id,
+                              longlong *func_value)
 {
   uint32 loc_part_id, sub_part_id;
   uint no_subparts;
-  longlong local_func_value;
   int error;
-  DBUG_ENTER("get_partition_id_range_sub_key");
+  DBUG_ENTER("get_partition_id_with_sub");
 
-  if (unlikely((error= get_partition_id_list(part_info, &loc_part_id,
-                                             func_value))))
+  if (unlikely((error= part_info->get_part_partition_id(part_info,
+                                                        &loc_part_id,
+                                                        func_value))))
   {
     DBUG_RETURN(error);
   }
   no_subparts= part_info->no_subparts;
-  sub_part_id= get_part_id_key(part_info->subpart_field_array,
-                               no_subparts, &local_func_value);
-  *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
-  DBUG_RETURN(0);
-}
-
-
-int get_partition_id_list_sub_linear_key(partition_info *part_info,
-                                          uint32 *part_id,
-                                          longlong *func_value)
-{
-  uint32 loc_part_id, sub_part_id;
-  uint no_subparts;
-  longlong local_func_value;
-  int error;
-  DBUG_ENTER("get_partition_id_list_sub_linear_key");
-
-  if (unlikely((error= get_partition_id_list(part_info, &loc_part_id,
-                                             func_value))))
+  if (unlikely((error= part_info->get_subpartition_id(part_info,
+                                                      &sub_part_id))))
   {
     DBUG_RETURN(error);
-  }
-  no_subparts= part_info->no_subparts;
-  sub_part_id= get_part_id_linear_key(part_info,
-                                      part_info->subpart_field_array,
-                                      no_subparts, &local_func_value);
+  } 
   *part_id= get_part_id_for_sub(loc_part_id, sub_part_id, no_subparts);
   DBUG_RETURN(0);
 }
@@ -4226,6 +4273,11 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
     partition_info *tab_part_info= table->part_info;
     partition_info *alt_part_info= thd->work_part_info;
     uint flags= 0;
+    bool is_last_partition_reorged;
+    part_elem_value *tab_max_elem_val= NULL;
+    part_elem_value *alt_max_elem_val= NULL;
+    longlong tab_max_range= 0, alt_max_range= 0;
+
     if (!tab_part_info)
     {
       my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
@@ -4280,34 +4332,43 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
       ((flags & (HA_FAST_CHANGE_PARTITION | HA_PARTITION_ONE_PHASE)) != 0);
     DBUG_PRINT("info", ("*fast_alter_partition: %d  flags: 0x%x",
                         *fast_alter_partition, flags));
-    if (((alter_info->flags & ALTER_ADD_PARTITION) ||
-         (alter_info->flags & ALTER_REORGANIZE_PARTITION)) &&
-         (thd->work_part_info->part_type != tab_part_info->part_type) &&
-         (thd->work_part_info->part_type != NOT_A_PARTITION))
+    if ((alter_info->flags & ALTER_ADD_PARTITION) ||
+         (alter_info->flags & ALTER_REORGANIZE_PARTITION))
     {
-      if (thd->work_part_info->part_type == RANGE_PARTITION)
-      {
-        my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
-                 "RANGE", "LESS THAN");
-      }
-      else if (thd->work_part_info->part_type == LIST_PARTITION)
+      if ((tab_part_info->column_list &&
+          alt_part_info->num_columns != tab_part_info->num_columns) ||
+          (!tab_part_info->column_list && alt_part_info->num_columns))
       {
-        DBUG_ASSERT(thd->work_part_info->part_type == LIST_PARTITION);
-        my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
-                 "LIST", "IN");
-      }
-      else if (tab_part_info->part_type == RANGE_PARTITION)
-      {
-        my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
-                 "RANGE", "LESS THAN");
+        my_error(ER_PARTITION_COLUMN_LIST_ERROR, MYF(0));
+        DBUG_RETURN(TRUE);
       }
-      else
+      if ((thd->work_part_info->part_type != tab_part_info->part_type) &&
+          (thd->work_part_info->part_type != NOT_A_PARTITION))
       {
-        DBUG_ASSERT(tab_part_info->part_type == LIST_PARTITION);
-        my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
-                 "LIST", "IN");
+        if (thd->work_part_info->part_type == RANGE_PARTITION)
+        {
+          my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
+                   "RANGE", "LESS THAN");
+        }
+        else if (thd->work_part_info->part_type == LIST_PARTITION)
+        {
+          DBUG_ASSERT(thd->work_part_info->part_type == LIST_PARTITION);
+          my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
+                   "LIST", "IN");
+        }
+        else if (tab_part_info->part_type == RANGE_PARTITION)
+        {
+          my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
+                   "RANGE", "LESS THAN");
+        }
+        else
+        {
+          DBUG_ASSERT(tab_part_info->part_type == LIST_PARTITION);
+          my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
+                   "LIST", "IN");
+        }
+        DBUG_RETURN(TRUE);
       }
-      DBUG_RETURN(TRUE);
     }
     if (alter_info->flags & ALTER_ADD_PARTITION)
     {
@@ -4747,7 +4808,6 @@ state of p1.
       */
       uint no_parts_reorged= alter_info->partition_names.elements;
       uint no_parts_new= thd->work_part_info->partitions.elements;
-      partition_info *alt_part_info= thd->work_part_info;
       uint check_total_partitions;
 
       tab_part_info->is_auto_partitioned= FALSE;
@@ -4827,9 +4887,7 @@ the generated partition syntax in a correct manner.
         uint part_count= 0;
         bool found_first= FALSE;
         bool found_last= FALSE;
-        bool is_last_partition_reorged;
         uint drop_count= 0;
-        longlong tab_max_range= 0, alt_max_range= 0;
         do
         {
           partition_element *part_elem= tab_it++;
@@ -4839,7 +4897,13 @@ the generated partition syntax in a correct manner.
           {
             is_last_partition_reorged= TRUE;
             drop_count++;
-            tab_max_range= part_elem->range_value;
+            if (tab_part_info->column_list)
+            {
+              List_iterator<part_elem_value> p(part_elem->list_val_list);
+              tab_max_elem_val= p++;
+            }
+            else
+              tab_max_range= part_elem->range_value;
             if (*fast_alter_partition &&
                 tab_part_info->temp_partitions.push_back(part_elem))
             {
@@ -4851,13 +4915,21 @@ the generated partition syntax in a correct manner.
             if (!found_first)
             {
               uint alt_part_count= 0;
-              found_first= TRUE;
+              partition_element *alt_part_elem;
               List_iterator<partition_element>
                                  alt_it(alt_part_info->partitions);
+              found_first= TRUE;
               do
               {
-                partition_element *alt_part_elem= alt_it++;
-                alt_max_range= alt_part_elem->range_value;
+                alt_part_elem= alt_it++;
+                if (tab_part_info->column_list)
+                {
+                  List_iterator<part_elem_value> p(alt_part_elem->list_val_list);
+                  alt_max_elem_val= p++;
+                }
+                else
+                  alt_max_range= alt_part_elem->range_value;
+
                 if (*fast_alter_partition)
                   alt_part_elem->part_state= PART_TO_BE_ADDED;
                 if (alt_part_count == 0)
@@ -4885,25 +4957,6 @@ the generated partition syntax in a correct manner.
           my_error(ER_DROP_PARTITION_NON_EXISTENT, MYF(0), "REORGANIZE");
           DBUG_RETURN(TRUE);
         }
-        if (tab_part_info->part_type == RANGE_PARTITION &&
-            ((is_last_partition_reorged &&
-               alt_max_range < tab_max_range) ||
-              (!is_last_partition_reorged &&
-               alt_max_range != tab_max_range)))
-        {
-          /*
-            For range partitioning the total resulting range before and
-            after the change must be the same except in one case. This is
-            when the last partition is reorganised, in this case it is
-            acceptable to increase the total range.
-            The reason is that it is not allowed to have "holes" in the
-            middle of the ranges and thus we should not allow to reorganise
-            to create "holes". Also we should not allow using REORGANIZE
-            to drop data.
-          */
-          my_error(ER_REORG_OUTSIDE_RANGE, MYF(0));
-          DBUG_RETURN(TRUE);
-        }
         tab_part_info->no_parts= check_total_partitions;
       }
     }
@@ -4923,8 +4976,40 @@ the generated partition syntax in a correct manner.
         tab_part_info->use_default_no_subpartitions= FALSE;
       }
       if (tab_part_info->check_partition_info(thd, (handlerton**)NULL,
-                                              table->file, ULL(0), FALSE))
+                                              table->file, ULL(0), TRUE))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      /*
+        The check below needs to be performed after check_partition_info
+        since this function "fixes" the item trees of the new partitions
+        to reorganize into
+      */
+      if (alter_info->flags == ALTER_REORGANIZE_PARTITION &&
+          tab_part_info->part_type == RANGE_PARTITION &&
+          ((is_last_partition_reorged &&
+            (tab_part_info->column_list ?
+             (tab_part_info->compare_column_values(
+                              alt_max_elem_val->col_val_array,
+                              tab_max_elem_val->col_val_array) < 0) :
+             alt_max_range < tab_max_range)) ||
+            (!is_last_partition_reorged &&
+             (tab_part_info->column_list ?
+              (tab_part_info->compare_column_values(
+                              alt_max_elem_val->col_val_array,
+                              tab_max_elem_val->col_val_array) != 0) :
+              alt_max_range != tab_max_range))))
       {
+        /*
+          For range partitioning the total resulting range before and
+          after the change must be the same except in one case. This is
+          when the last partition is reorganised, in this case it is
+          acceptable to increase the total range.
+          The reason is that it is not allowed to have "holes" in the
+          middle of the ranges and thus we should not allow to reorganise
+          to create "holes".
+        */
+        my_error(ER_REORG_OUTSIDE_RANGE, MYF(0));
         DBUG_RETURN(TRUE);
       }
     }
@@ -6565,16 +6650,19 @@ void make_used_partitions_str(partition_info *part_info, String *parts_str)
   IMPLEMENTATION
     There are two available interval analyzer functions:
     (1) get_part_iter_for_interval_via_mapping 
-    (2) get_part_iter_for_interval_via_walking
+    (2) get_part_iter_for_interval_cols_via_map 
+    (3) get_part_iter_for_interval_via_walking
 
     They both have limited applicability:
     (1) is applicable for "PARTITION BY <RANGE|LIST>(func(t.field))", where
     func is a monotonic function.
-    
-    (2) is applicable for 
+
+    (2) is applicable for "PARTITION BY <RANGE|LIST> COLUMN_LIST (field_list)
+
+    (3) is applicable for 
       "[SUB]PARTITION BY <any-partitioning-type>(any_func(t.integer_field))"
       
-    If both are applicable, (1) is preferred over (2).
+    If both (1) and (3) are applicable, (1) is preferred over (3).
     
     This function sets part_info::get_part_iter_for_interval according to
     this criteria, and also sets some auxilary fields that the function
@@ -6594,10 +6682,19 @@ static void set_up_range_analysis_info(partition_info *part_info)
   switch (part_info->part_type) {
   case RANGE_PARTITION:
   case LIST_PARTITION:
-    if (part_info->part_expr->get_monotonicity_info() != NON_MONOTONIC)
+    if (!part_info->column_list)
+    {
+      if (part_info->part_expr->get_monotonicity_info() != NON_MONOTONIC)
+      {
+        part_info->get_part_iter_for_interval=
+          get_part_iter_for_interval_via_mapping;
+        goto setup_subparts;
+      }
+    }
+    else
     {
       part_info->get_part_iter_for_interval=
-        get_part_iter_for_interval_via_mapping;
+        get_part_iter_for_interval_cols_via_map;
       goto setup_subparts;
     }
   default:
@@ -6648,9 +6745,104 @@ setup_subparts:
 }
 
 
+/* TODO Commenting those functions */
+uint32 store_tuple_to_record(Field **pfield,
+                             uint32 *store_length_array,
+                             uchar *value,
+                             uchar *value_end)
+{
+  // see store_key_image_to_rec
+  uint32 nparts= 0;
+  uchar *loc_value;
+  while (value < value_end)
+  {
+    loc_value= value;
+    if ((*pfield)->real_maybe_null())
+    {
+      if (*loc_value)
+      {
+        (*pfield)->set_null();
+      }
+      (*pfield)->set_notnull();
+      loc_value++;
+    }
+    uint len= (*pfield)->pack_length();
+    (*pfield)->set_key_image(loc_value, len);
+    value+= *store_length_array;
+    store_length_array++;
+    nparts++;
+    pfield++;
+  }
+  return nparts;
+}
+
+/*
+  RANGE(columns) partitioning: compare value bound and probe tuple.
+
+  The value bound always is a full tuple (but may include MIN_VALUE and
+  MAX_VALUE special values).
+
+  The probe tuple may be a prefix of partitioning tuple. The tail_is_min
+  parameter specifies whether the suffix components should be assumed to
+  hold MIN_VALUE or MAX_VALUE
+*/
+
+static int cmp_rec_and_tuple(part_column_list_val *val, uint32 nvals_in_rec)
+{
+  partition_info *part_info= val->part_info;
+  Field **field= part_info->part_field_array;
+  Field **fields_end= field + nvals_in_rec;
+  int res;
+
+  for (; field != fields_end; field++, val++)
+  {
+    if (val->max_value)
+      return -1;
+    if ((*field)->is_null())
+    {
+      if (val->null_value)
+        continue;
+      return -1;
+    }
+    if (val->null_value)
+      return +1;
+    res= (*field)->cmp((const uchar*)val->column_value);
+    if (res)
+      return res;
+  }
+  return 0;
+}
+
+
+static int cmp_rec_and_tuple_prune(part_column_list_val *val,
+                                   uint32 n_vals_in_rec,
+                                   bool tail_is_min)
+{
+  int cmp;
+  Field **field;
+  partition_info *part_info;
+  if ((cmp= cmp_rec_and_tuple(val, n_vals_in_rec)))
+    return cmp;
+  part_info= val->part_info;
+  field= part_info->part_field_array + n_vals_in_rec;
+  for (; *field; field++, val++)
+  {
+    if (tail_is_min)
+      return -1;
+    if (!tail_is_min && !val->max_value)
+      return +1;
+  }
+  return 0;
+}
+
+
 typedef uint32 (*get_endpoint_func)(partition_info*, bool left_endpoint,
                                     bool include_endpoint);
 
+typedef uint32 (*get_col_endpoint_func)(partition_info*, bool left_endpoint,
+                                        bool include_endpoint,
+                                        uint32 no_parts);
+
 /*
   Partitioning Interval Analysis: Initialize the iterator for "mapping" case
 
@@ -6686,17 +6878,132 @@ typedef uint32 (*get_endpoint_func)(partition_info*, bool left_endpoint,
    -1 - All partitions would match (iterator not initialized)
 */
 
+uint32 get_partition_id_cols_range_for_endpoint(partition_info *part_info,
+                                                bool left_endpoint,
+                                                bool include_endpoint,
+                                                uint32 nparts)
+{
+  uint max_partition= part_info->no_parts - 1;
+  uint min_part_id= 0, max_part_id= max_partition, loc_part_id;
+  part_column_list_val *range_col_array= part_info->range_col_array;
+  uint no_columns= part_info->part_field_list.elements;
+  bool tailf= !(left_endpoint ^ include_endpoint);
+  DBUG_ENTER("get_partition_id_cols_range_for_endpoint");
+
+  /* Get the partitioning function value for the endpoint */
+  while (max_part_id > min_part_id)
+  {
+    loc_part_id= (max_part_id + min_part_id + 1) >> 1;
+    if (cmp_rec_and_tuple_prune(range_col_array + loc_part_id*no_columns,
+                                nparts, tailf) >= 0)
+      min_part_id= loc_part_id + 1;
+    else
+      max_part_id= loc_part_id - 1;
+  }
+  loc_part_id= max_part_id;
+  if (loc_part_id < max_partition && 
+      cmp_rec_and_tuple_prune(range_col_array + (loc_part_id+1)*no_columns,
+                              nparts, tailf) >= 0
+      )
+  {
+     loc_part_id++;
+  }
+  if (left_endpoint)
+  {
+    if (cmp_rec_and_tuple_prune(range_col_array + loc_part_id*no_columns,
+                                nparts, tailf) >= 0)
+      loc_part_id++;
+  }
+  else 
+  {
+    if (loc_part_id < max_partition)
+    {
+      int res= cmp_rec_and_tuple_prune(range_col_array +
+                                       loc_part_id * no_columns,
+                                       nparts, tailf);
+      if (!res)
+        loc_part_id += test(include_endpoint);
+      else if (res > 0)
+        loc_part_id++;
+    }
+    loc_part_id++;
+  }
+  DBUG_RETURN(loc_part_id);
+}
+
+
+int get_part_iter_for_interval_cols_via_map(partition_info *part_info,
+                                            bool is_subpart,
+                                            uint32 *store_length_array,
+                                            uchar *min_value, uchar *max_value,
+                                            uint min_len, uint max_len, 
+                                            uint flags,
+                                            PARTITION_ITERATOR *part_iter)
+{
+  uint32 nparts;
+  get_col_endpoint_func  get_col_endpoint;
+  DBUG_ENTER("get_part_iter_for_interval_cols_via_map");
+
+  if (part_info->part_type == RANGE_PARTITION)
+  {
+    get_col_endpoint= get_partition_id_cols_range_for_endpoint;
+    part_iter->get_next= get_next_partition_id_range;
+  }
+  else
+  {
+    get_col_endpoint= get_partition_id_cols_list_for_endpoint;
+    part_iter->get_next= get_next_partition_id_list;
+    part_iter->part_info= part_info;
+  }
+  if (flags & NO_MIN_RANGE)
+    part_iter->part_nums.start= part_iter->part_nums.cur= 0;
+  else
+  {
+    // Copy from min_value to record
+    nparts= store_tuple_to_record(part_info->part_field_array,
+                                  store_length_array,
+                                  min_value,
+                                  min_value + min_len);
+    part_iter->part_nums.start= part_iter->part_nums.cur=
+      get_col_endpoint(part_info, TRUE, !(flags & NEAR_MIN),
+                       nparts);
+  }
+  if (flags & NO_MAX_RANGE)
+    part_iter->part_nums.end= part_info->no_parts;
+  else
+  {
+    // Copy from max_value to record
+    nparts= store_tuple_to_record(part_info->part_field_array,
+                                  store_length_array,
+                                  max_value,
+                                  max_value + max_len);
+    part_iter->part_nums.end= get_col_endpoint(part_info, FALSE,
+                                               !(flags & NEAR_MAX),
+                                               nparts);
+  }
+  if (part_iter->part_nums.start == part_iter->part_nums.end)
+    DBUG_RETURN(0);
+  DBUG_RETURN(1);
+}
+
+
 int get_part_iter_for_interval_via_mapping(partition_info *part_info,
                                            bool is_subpart,
+                                           uint32 *store_length_array, /* ignored */
                                            uchar *min_value, uchar *max_value,
+                                           uint min_len, uint max_len, /* ignored */
                                            uint flags,
                                            PARTITION_ITERATOR *part_iter)
 {
-  DBUG_ASSERT(!is_subpart);
   Field *field= part_info->part_field_array[0];
   uint32             max_endpoint_val;
   get_endpoint_func  get_endpoint;
   uint field_len= field->pack_length_in_rec();
+  DBUG_ENTER("get_part_iter_for_interval_via_mapping");
+  DBUG_ASSERT(!is_subpart);
+  (void) store_length_array;
+  (void)min_len;
+  (void)max_len;
   part_iter->ret_null_part= part_iter->ret_null_part_orig= FALSE;
 
   if (part_info->part_type == RANGE_PARTITION)
@@ -6728,7 +7035,7 @@ int get_part_iter_for_interval_via_mapping(partition_info *part_info,
       part_iter->part_nums.start= part_iter->part_nums.end= 0;
       part_iter->part_nums.cur= 0;
       part_iter->ret_null_part= part_iter->ret_null_part_orig= TRUE;
-      return -1;
+      DBUG_RETURN(-1);
     }
   }
   else
@@ -6747,7 +7054,7 @@ int get_part_iter_for_interval_via_mapping(partition_info *part_info,
     {
       /* The right bound is X <= NULL, i.e. it is a "X IS NULL" interval */
       part_iter->part_nums.end= 0;
-      return 1;
+      DBUG_RETURN(1);
     }
   }
   else
@@ -6767,7 +7074,7 @@ int get_part_iter_for_interval_via_mapping(partition_info *part_info,
       part_iter->part_nums.start= get_endpoint(part_info, 1, include_endp);
       part_iter->part_nums.cur= part_iter->part_nums.start;
       if (part_iter->part_nums.start == max_endpoint_val)
-        return 0; /* No partitions */
+        DBUG_RETURN(0); /* No partitions */
     }
   }
 
@@ -6781,9 +7088,9 @@ int get_part_iter_for_interval_via_mapping(partition_info *part_info,
     part_iter->part_nums.end= get_endpoint(part_info, 0, include_endp);
     if (part_iter->part_nums.start >= part_iter->part_nums.end &&
         !part_iter->ret_null_part)
-      return 0; /* No partitions */
+      DBUG_RETURN(0); /* No partitions */
   }
-  return 1; /* Ok, iterator initialized */
+  DBUG_RETURN(1); /* Ok, iterator initialized */
 }
 
 
@@ -6841,14 +7148,21 @@ int get_part_iter_for_interval_via_mapping(partition_info *part_info,
 */
 
 int get_part_iter_for_interval_via_walking(partition_info *part_info,
-                                           bool is_subpart,
-                                           uchar *min_value, uchar *max_value,
-                                           uint flags,
-                                           PARTITION_ITERATOR *part_iter)
+                                      bool is_subpart,
+                                      uint32 *store_length_array, /* ignored */
+                                      uchar *min_value, uchar *max_value,
+                                      uint min_len, uint max_len, /* ignored */
+                                      uint flags,
+                                      PARTITION_ITERATOR *part_iter)
 {
   Field *field;
   uint total_parts;
   partition_iter_func get_next_func;
+  DBUG_ENTER("get_part_iter_for_interval_via_walking");
+  (void)store_length_array;
+  (void)min_len;
+  (void)max_len;
+
   if (is_subpart)
   {
     field= part_info->subpart_field_array[0];
@@ -6878,7 +7192,7 @@ int get_part_iter_for_interval_via_walking(partition_info *part_info,
       if (!part_info->get_subpartition_id(part_info, &part_id))
       {
         init_single_partition_iterator(part_id, part_iter);
-        return 1; /* Ok, iterator initialized */
+        DBUG_RETURN(1); /* Ok, iterator initialized */
       }
     }
     else
@@ -6891,10 +7205,10 @@ int get_part_iter_for_interval_via_walking(partition_info *part_info,
       if (!res)
       {
         init_single_partition_iterator(part_id, part_iter);
-        return 1; /* Ok, iterator initialized */
+        DBUG_RETURN(1); /* Ok, iterator initialized */
       }
     }
-    return 0; /* No partitions match */
+    DBUG_RETURN(0); /* No partitions match */
   }
 
   if ((field->real_maybe_null() && 
@@ -6902,7 +7216,7 @@ int get_part_iter_for_interval_via_walking(partition_info *part_info,
         (!(flags & NO_MAX_RANGE) && *max_value))) ||  // X <? NULL
       (flags & (NO_MIN_RANGE | NO_MAX_RANGE)))    // -inf at any bound
   {
-    return -1; /* Can't handle this interval, have to use all partitions */
+    DBUG_RETURN(-1); /* Can't handle this interval, have to use all partitions */
   }
   
   /* Get integers for left and right interval bound */
@@ -6921,20 +7235,20 @@ int get_part_iter_for_interval_via_walking(partition_info *part_info,
     an empty interval by "wrapping around" a + 4G-1 + 1 = a. 
   */
   if ((ulonglong)b - (ulonglong)a == ~0ULL)
-    return -1;
+    DBUG_RETURN(-1);
 
   a += test(flags & NEAR_MIN);
   b += test(!(flags & NEAR_MAX));
   ulonglong n_values= b - a;
   
   if (n_values > total_parts || n_values > MAX_RANGE_TO_WALK)
-    return -1;
+    DBUG_RETURN(-1);
 
   part_iter->field_vals.start= part_iter->field_vals.cur= a;
   part_iter->field_vals.end=   b;
   part_iter->part_info= part_info;
   part_iter->get_next=  get_next_func;
-  return 1;
+  DBUG_RETURN(1);
 }
 
 
@@ -6976,8 +7290,9 @@ uint32 get_next_partition_id_range(PARTITION_ITERATOR* part_iter)
 
   DESCRIPTION
     This implementation of PARTITION_ITERATOR::get_next() is special for 
-    LIST partitioning: it enumerates partition ids in 
-    part_info->list_array[i] where i runs over [min_idx, max_idx] interval.
+    LIST partitioning: it enumerates partition ids in
+    part_info->list_array[i] (list_col_array[i] for COLUMN_LIST LIST
+    partitioning) where i runs over [min_idx, max_idx] interval.
     The function conforms to partition_iter_func type.
 
   RETURN 
@@ -6999,8 +7314,13 @@ uint32 get_next_partition_id_list(PARTITION_ITERATOR *part_iter)
     return NOT_A_PARTITION_ID;
   }
   else
-    return part_iter->part_info->list_array[part_iter->
-                                            part_nums.cur++].partition_id;
+  {
+    partition_info *part_info= part_iter->part_info;
+    uint32 num_part= part_iter->part_nums.cur++;
+    return part_info->column_list ?
+           part_info->list_col_array[num_part].partition_id :
+           part_info->list_array[num_part].partition_id;
+  }
 }
 
 
diff --git a/sql/sql_partition.h b/sql/sql_partition.h
index 282e24f1853..7902cc77877 100644
--- a/sql/sql_partition.h
+++ b/sql/sql_partition.h
@@ -173,9 +173,12 @@ typedef struct st_partition_iter
   SYNOPSIS
     get_partitions_in_range_iter()
       part_info   Partitioning info
-      is_subpart  
+      is_subpart
+      store_length_array Length of fields packed in opt_range_key format
       min_val     Left edge,  field value in opt_range_key format.
-      max_val     Right edge, field value in opt_range_key format. 
+      max_val     Right edge, field value in opt_range_key format.
+      min_len     Length of minimum value
+      max_len     Length of maximum value
       flags       Some combination of NEAR_MIN, NEAR_MAX, NO_MIN_RANGE,
                   NO_MAX_RANGE.
       part_iter   Iterator structure to be initialized
@@ -191,8 +194,9 @@ typedef struct st_partition_iter
     The set of partitions is returned by initializing an iterator in *part_iter
 
   NOTES
-    There are currently two functions of this type:
+    There are currently three functions of this type:
      - get_part_iter_for_interval_via_walking
+     - get_part_iter_for_interval_cols_via_map
      - get_part_iter_for_interval_via_mapping
 
   RETURN 
@@ -203,7 +207,9 @@ typedef struct st_partition_iter
 
 typedef int (*get_partitions_in_range_iter)(partition_info *part_info,
                                             bool is_subpart,
+                                            uint32 *store_length_array,
                                             uchar *min_val, uchar *max_val,
+                                            uint min_len, uint max_len,
                                             uint flags,
                                             PARTITION_ITERATOR *part_iter);
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 5c2c351652b..3aef62dfb31 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -4869,12 +4869,18 @@ static int get_schema_partitions_record(THD *thd, TABLE_LIST *tables,
     /* Partition method*/
     switch (part_info->part_type) {
     case RANGE_PARTITION:
-      table->field[7]->store(partition_keywords[PKW_RANGE].str,
-                             partition_keywords[PKW_RANGE].length, cs);
-      break;
     case LIST_PARTITION:
-      table->field[7]->store(partition_keywords[PKW_LIST].str,
-                             partition_keywords[PKW_LIST].length, cs);
+      tmp_res.length(0);
+      if (part_info->part_type == RANGE_PARTITION)
+        tmp_res.append(partition_keywords[PKW_RANGE].str,
+                       partition_keywords[PKW_RANGE].length);
+      else
+        tmp_res.append(partition_keywords[PKW_LIST].str,
+                       partition_keywords[PKW_LIST].length);
+      if (part_info->column_list)
+        tmp_res.append(partition_keywords[PKW_COLUMNS].str,
+                       partition_keywords[PKW_COLUMNS].length);
+      table->field[7]->store(tmp_res.ptr(), tmp_res.length(), cs);
       break;
     case HASH_PARTITION:
       tmp_res.length(0);
@@ -6484,7 +6490,7 @@ ST_FIELD_INFO partitions_fields_info[]=
    (MY_I_S_MAYBE_NULL | MY_I_S_UNSIGNED), 0, OPEN_FULL_TABLE},
   {"SUBPARTITION_ORDINAL_POSITION", 21 , MYSQL_TYPE_LONGLONG, 0,
    (MY_I_S_MAYBE_NULL | MY_I_S_UNSIGNED), 0, OPEN_FULL_TABLE},
-  {"PARTITION_METHOD", 12, MYSQL_TYPE_STRING, 0, 1, 0, OPEN_FULL_TABLE},
+  {"PARTITION_METHOD", 18, MYSQL_TYPE_STRING, 0, 1, 0, OPEN_FULL_TABLE},
   {"SUBPARTITION_METHOD", 12, MYSQL_TYPE_STRING, 0, 1, 0, OPEN_FULL_TABLE},
   {"PARTITION_EXPRESSION", 65535, MYSQL_TYPE_STRING, 0, 1, 0, OPEN_FULL_TABLE},
   {"SUBPARTITION_EXPRESSION", 65535, MYSQL_TYPE_STRING, 0, 1, 0,
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 81d00f46000..fce51c7e3da 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3664,7 +3664,7 @@ bool mysql_create_table_no_lock(THD *thd,
              ha_resolve_storage_engine_name(part_info->default_engine_type),
              ha_resolve_storage_engine_name(create_info->db_type)));
     if (part_info->check_partition_info(thd, &engine_type, file,
-                                        create_info, TRUE))
+                                        create_info, FALSE))
       goto err;
     part_info->default_engine_type= engine_type;
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 4ed9946a334..0e0e3ec6bd3 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -515,10 +515,10 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 
 %pure_parser                                    /* We have threads */
 /*
-  Currently there are 169 shift/reduce conflicts.
+  Currently there are 168 shift/reduce conflicts.
   We should not introduce new conflicts any more.
 */
-%expect 169
+%expect 168
 
 /*
    Comments for TOKENS.
@@ -603,6 +603,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %token  COLLATE_SYM                   /* SQL-2003-R */
 %token  COLLATION_SYM                 /* SQL-2003-N */
 %token  COLUMNS
+%token  COLUMN_LIST_SYM
 %token  COLUMN_SYM                    /* SQL-2003-R */
 %token  COMMENT_SYM
 %token  COMMITTED_SYM                 /* SQL-2003-N */
@@ -1156,6 +1157,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         opt_natural_language_mode opt_query_expansion
         opt_ev_status opt_ev_on_completion ev_on_completion opt_ev_comment
         ev_alter_on_schedule_completion opt_ev_rename_to opt_ev_sql_stmt
+        opt_global
 
 %type <ulong_num>
         ulong_num real_ulong_num merge_insert_types
@@ -1298,6 +1300,8 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         view_check_option trigger_tail sp_tail sf_tail udf_tail event_tail
         install uninstall partition_entry binlog_base64_event
         init_key_options key_options key_opts key_opt key_using_alg
+        part_column_list part_column_expr_list part_column_expr_item
+        part_column_list_value
         server_def server_options_list server_option
         definer_opt no_definer definer
 END_OF_INPUT
@@ -1691,12 +1695,12 @@ create:
                                   $5->table.str);
             }
           }
-        | CREATE opt_unique_or_fulltext INDEX_SYM ident key_alg ON
+        | CREATE opt_global opt_unique_or_fulltext INDEX_SYM ident key_alg ON
           table_ident
           {
             LEX *lex=Lex;
             lex->sql_command= SQLCOM_CREATE_INDEX;
-            if (!lex->current_select->add_table_to_list(lex->thd, $7,
+            if (!lex->current_select->add_table_to_list(lex->thd, $8,
                                                         NULL,
                                                         TL_OPTION_UPDATING))
               MYSQL_YYABORT;
@@ -1704,6 +1708,7 @@ create:
             lex->alter_info.flags= ALTER_ADD_INDEX;
             lex->col_list.empty();
             lex->change=NullS;
+            lex->global_flag= $2;
           }
           '(' key_list ')' key_options
           {
@@ -1714,13 +1719,22 @@ create:
               my_parse_error(ER(ER_SYNTAX_ERROR));
               MYSQL_YYABORT;
             }
-            key= new Key($2, $4.str, &lex->key_create_info, 0,
+            key= new Key($3, $5.str, &lex->key_create_info, 0,
                          lex->col_list);
             if (key == NULL)
               MYSQL_YYABORT;
             lex->alter_info.key_list.push_back(key);
             lex->col_list.empty();
           }
+          opt_partitioning
+          {
+            LEX *lex= Lex;
+            if (!lex->global_flag && lex->part_info)
+            {
+              my_error(ER_GLOBAL_PARTITION_INDEX_ERROR, MYF(0));
+              YYABORT;
+            }
+          }
         | CREATE DATABASE opt_if_not_exists ident
           {
             Lex->create_info.default_table_charset= NULL;
@@ -3802,19 +3816,21 @@ partition:
 part_type_def:
           opt_linear KEY_SYM '(' part_field_list ')'
           {
-            LEX *lex= Lex;
-            lex->part_info->list_of_part_fields= TRUE;
-            lex->part_info->part_type= HASH_PARTITION;
+            partition_info *part_info= Lex->part_info;
+            part_info->list_of_part_fields= TRUE;
+            part_info->part_type= HASH_PARTITION;
           }
         | opt_linear HASH_SYM
           { Lex->part_info->part_type= HASH_PARTITION; }
           part_func {}
-        | RANGE_SYM
+        | RANGE_SYM part_func
           { Lex->part_info->part_type= RANGE_PARTITION; }
-          part_func {}
-        | LIST_SYM
+        | RANGE_SYM part_column_list
+          { Lex->part_info->part_type= RANGE_PARTITION; }
+        | LIST_SYM part_func
+          { Lex->part_info->part_type= LIST_PARTITION; }
+        | LIST_SYM part_column_list
           { Lex->part_info->part_type= LIST_PARTITION; }
-          part_func {}
         ;
 
 opt_linear:
@@ -3836,41 +3852,45 @@ part_field_item_list:
 part_field_item:
           ident
           {
-            if (Lex->part_info->part_field_list.push_back($1.str))
+            partition_info *part_info= Lex->part_info;
+            part_info->num_columns++;
+            if (part_info->part_field_list.push_back($1.str))
             {
               mem_alloc_error(1);
               MYSQL_YYABORT;
             }
+            if (part_info->num_columns > MAX_REF_PARTS)
+            {
+              my_error(ER_TOO_MANY_PARTITION_FUNC_FIELDS_ERROR, MYF(0),
+                       "list of partition fields");
+              MYSQL_YYABORT;
+            }
           }
         ;
 
+part_column_list:
+        COLUMN_LIST_SYM '(' part_field_list ')'
+        {
+          partition_info *part_info= Lex->part_info;
+          part_info->column_list= TRUE;
+          part_info->list_of_part_fields= TRUE;
+        }
+        ;
+
+
 part_func:
           '(' remember_name part_func_expr remember_end ')'
           {
-            LEX *lex= Lex;
-            uint expr_len= (uint)($4 - $2) - 1;
-            lex->part_info->list_of_part_fields= FALSE;
-            lex->part_info->part_expr= $3;
-            char *func_string= (char*) sql_memdup($2+1, expr_len);
-            if (func_string == NULL)
-              MYSQL_YYABORT;
-            lex->part_info->part_func_string= func_string;
-            lex->part_info->part_func_len= expr_len;
+            if (Lex->part_info->set_part_expr($2+1, $3, $4, FALSE))
+            { MYSQL_YYABORT; }
           }
         ;
 
 sub_part_func:
           '(' remember_name part_func_expr remember_end ')'
           {
-            LEX *lex= Lex;
-            uint expr_len= (uint)($4 - $2) - 1;
-            lex->part_info->list_of_subpart_fields= FALSE;
-            lex->part_info->subpart_expr= $3;
-            char *func_string= (char*) sql_memdup($2+1, expr_len);
-            if (func_string == NULL)
-              MYSQL_YYABORT;
-            lex->part_info->subpart_func_string= func_string;        
-            lex->part_info->subpart_func_len= expr_len;
+            if (Lex->part_info->set_part_expr($2+1, $3, $4, TRUE))
+            { MYSQL_YYABORT; }
           }
         ;
 
@@ -3880,15 +3900,15 @@ opt_no_parts:
         | PARTITIONS_SYM real_ulong_num
           { 
             uint no_parts= $2;
-            LEX *lex= Lex;
+            partition_info *part_info= Lex->part_info;
             if (no_parts == 0)
             {
               my_error(ER_NO_PARTS_ERROR, MYF(0), "partitions");
               MYSQL_YYABORT;
             }
 
-            lex->part_info->no_parts= no_parts;
-            lex->part_info->use_default_no_partitions= FALSE;
+            part_info->no_parts= no_parts;
+            part_info->use_default_no_partitions= FALSE;
           }
         ;
 
@@ -3900,9 +3920,9 @@ opt_sub_part:
         | SUBPARTITION_SYM BY opt_linear KEY_SYM
           '(' sub_part_field_list ')'
           {
-            LEX *lex= Lex;
-            lex->part_info->subpart_type= HASH_PARTITION;
-            lex->part_info->list_of_subpart_fields= TRUE;
+            partition_info *part_info= Lex->part_info;
+            part_info->subpart_type= HASH_PARTITION;
+            part_info->list_of_subpart_fields= TRUE;
           }
           opt_no_subparts {}
         ;
@@ -3915,11 +3935,18 @@ sub_part_field_list:
 sub_part_field_item:
           ident
           {
-            if (Lex->part_info->subpart_field_list.push_back($1.str))
+            partition_info *part_info= Lex->part_info;
+            if (part_info->subpart_field_list.push_back($1.str))
             {
               mem_alloc_error(1);
               MYSQL_YYABORT;
             }
+            if (part_info->subpart_field_list.elements > MAX_REF_PARTS)
+            {
+              my_error(ER_TOO_MANY_PARTITION_FUNC_FIELDS_ERROR, MYF(0),
+                       "list of subpartition fields");
+              MYSQL_YYABORT;
+            }
           }
         ;
 
@@ -3960,8 +3987,7 @@ part_defs:
           {}
         | '(' part_def_list ')'
           {
-            LEX *lex= Lex;
-            partition_info *part_info= lex->part_info;
+            partition_info *part_info= Lex->part_info;
             uint count_curr_parts= part_info->partitions.elements;
             if (part_info->no_parts != 0)
             {
@@ -3988,8 +4014,7 @@ part_def_list:
 part_definition:
           PARTITION_SYM
           {
-            LEX *lex= Lex;
-            partition_info *part_info= lex->part_info;
+            partition_info *part_info= Lex->part_info;
             partition_element *p_elem= new partition_element();
 
             if (!p_elem || part_info->partitions.push_back(p_elem))
@@ -4013,8 +4038,7 @@ part_definition:
 part_name:
           ident
           {
-            LEX *lex= Lex;
-            partition_info *part_info= lex->part_info;
+            partition_info *part_info= Lex->part_info;
             partition_element *p_elem= part_info->curr_part_elem;
             p_elem->partition_name= $1.str;
           }
@@ -4024,15 +4048,16 @@ opt_part_values:
           /* empty */
           {
             LEX *lex= Lex;
+            partition_info *part_info= lex->part_info;
             if (! lex->is_partition_management())
             {
-              if (lex->part_info->part_type == RANGE_PARTITION)
+              if (part_info->part_type == RANGE_PARTITION)
               {
                 my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
                          "RANGE", "LESS THAN");
                 MYSQL_YYABORT;
               }
-              if (lex->part_info->part_type == LIST_PARTITION)
+              if (part_info->part_type == LIST_PARTITION)
               {
                 my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
                          "LIST", "IN");
@@ -4040,14 +4065,15 @@ opt_part_values:
               }
             }
             else
-              lex->part_info->part_type= HASH_PARTITION;
+              part_info->part_type= HASH_PARTITION;
           }
         | VALUES LESS_SYM THAN_SYM part_func_max
           {
             LEX *lex= Lex;
+            partition_info *part_info= lex->part_info;
             if (! lex->is_partition_management())
             {
-              if (Lex->part_info->part_type != RANGE_PARTITION)
+              if (part_info->part_type != RANGE_PARTITION)
               {
                 my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
                          "RANGE", "LESS THAN");
@@ -4055,14 +4081,15 @@ opt_part_values:
               }
             }
             else
-              lex->part_info->part_type= RANGE_PARTITION;
+              part_info->part_type= RANGE_PARTITION;
           }
         | VALUES IN_SYM '(' part_list_func ')'
           {
             LEX *lex= Lex;
+            partition_info *part_info= lex->part_info;
             if (! lex->is_partition_management())
             {
-              if (Lex->part_info->part_type != LIST_PARTITION)
+              if (part_info->part_type != LIST_PARTITION)
               {
                 my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
                          "LIST", "IN");
@@ -4070,36 +4097,139 @@ opt_part_values:
               }
             }
             else
-              lex->part_info->part_type= LIST_PARTITION;
+              part_info->part_type= LIST_PARTITION;
+          }
+        ;
+
+part_column_expr_list:
+          part_column_expr_item {}
+        | part_column_expr_list ',' part_column_expr_item {}
+        ;
+
+part_column_expr_item:
+          MAX_VALUE_SYM
+          {
+            partition_info *part_info= Lex->part_info;
+            part_column_list_val *col_val;
+            if (part_info->part_type == LIST_PARTITION)
+            {
+              my_parse_error(ER(ER_MAXVALUE_IN_LIST_PARTITIONING_ERROR));
+              MYSQL_YYABORT;
+            }
+            if (!(col_val= part_info->add_column_value()))
+            {
+              MYSQL_YYABORT;
+            }
+            col_val->max_value= TRUE;
+          }
+        | bit_expr
+          {
+            part_column_list_val *col_val;
+            LEX *lex= Lex;
+            if (!lex->safe_to_cache_query)
+            {
+              my_error(ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR, MYF(0));
+              MYSQL_YYABORT;
+            }
+            if (!(col_val= lex->part_info->add_column_value()))
+            {
+              MYSQL_YYABORT;
+            }
+            col_val->item_expression= $1;
+            col_val->part_info= NULL;
+          }
+        ;
+
+part_column_list_value:
+          COLUMN_LIST_SYM
+          {
+            LEX *lex= Lex;
+            partition_info *part_info= lex->part_info;
+            uint num_columns;
+            partition_element *p_elem= part_info->curr_part_elem;
+            part_column_list_val *col_val_array;
+            part_elem_value *list_val;
+
+            if (!part_info->column_list &&
+                !lex->is_partition_management())
+            {
+              my_error(ER_PARTITION_COLUMN_LIST_ERROR, MYF(0));
+              MYSQL_YYABORT;
+            }
+            if (!(list_val= 
+                 (part_elem_value*)sql_calloc(sizeof(part_elem_value))) ||
+                p_elem->list_val_list.push_back(list_val))
+            {
+              mem_alloc_error(sizeof(part_elem_value));
+              MYSQL_YYABORT;
+            }
+            if (part_info->num_columns)
+              num_columns= part_info->num_columns;
+            else
+              num_columns= MAX_REF_PARTS;
+            if (!(col_val_array=
+                (part_column_list_val*)sql_calloc(num_columns *
+                                       sizeof(part_column_list_val))))
+            {
+              mem_alloc_error(num_columns * sizeof(part_elem_value));
+              MYSQL_YYABORT;
+            }
+            list_val->col_val_array= col_val_array;
+            part_info->curr_list_val= list_val;
+            part_info->curr_list_object= 0;
+          }
+          '(' part_column_expr_list ')'
+          {
+            partition_info *part_info= Lex->part_info;
+            uint num_columns= part_info->num_columns;
+            if (num_columns && num_columns != part_info->curr_list_object)
+            {
+              my_error(ER_PARTITION_COLUMN_LIST_ERROR, MYF(0));
+              MYSQL_YYABORT;
+            }
+            part_info->num_columns= part_info->curr_list_object;
           }
         ;
 
 part_func_max:
           max_value_sym
           {
-            LEX *lex= Lex;
-            if (lex->part_info->defined_max_value)
+            partition_info *part_info= Lex->part_info;
+            if (part_info->defined_max_value)
             {
               my_parse_error(ER(ER_PARTITION_MAXVALUE_ERROR));
               MYSQL_YYABORT;
             }
-            lex->part_info->defined_max_value= TRUE;
-            lex->part_info->curr_part_elem->max_value= TRUE;
-            lex->part_info->curr_part_elem->range_value= LONGLONG_MAX;
+            if (part_info->column_list)
+            {
+              my_parse_error(ER(ER_PARTITION_COLUMN_LIST_ERROR));
+              MYSQL_YYABORT;
+            }
+            part_info->defined_max_value= TRUE;
+            part_info->curr_part_elem->max_value= TRUE;
+            part_info->curr_part_elem->range_value= LONGLONG_MAX;
           }
         | part_range_func
           {
-            if (Lex->part_info->defined_max_value)
+            partition_info *part_info= Lex->part_info;
+            if (part_info->defined_max_value)
             {
               my_parse_error(ER(ER_PARTITION_MAXVALUE_ERROR));
               MYSQL_YYABORT;
             }
-            if (Lex->part_info->curr_part_elem->has_null_value)
+            if (part_info->curr_part_elem->has_null_value)
             {
               my_parse_error(ER(ER_NULL_IN_VALUES_LESS_THAN));
               MYSQL_YYABORT;
             }
+            if (part_info->column_list)
+            {
+              my_parse_error(ER(ER_PARTITION_COLUMN_LIST_ERROR));
+              MYSQL_YYABORT;
+            }
           }
+        | '(' part_column_list_value ')'
+          {}
         ;
 
 max_value_sym:
@@ -4136,7 +4266,13 @@ part_list_item:
               mem_alloc_error(sizeof(part_elem_value));
               MYSQL_YYABORT;
             }
+            if (part_info->column_list)
+            {
+              my_parse_error(ER(ER_PARTITION_COLUMN_LIST_ERROR));
+              MYSQL_YYABORT;
+            }
           }
+        | part_column_list_value
         ;
 
 part_bit_expr:
@@ -4145,6 +4281,7 @@ part_bit_expr:
             Item *part_expr= $1;
             THD *thd= YYTHD;
             LEX *lex= thd->lex;
+            partition_info *part_info= lex->part_info;
             Name_resolution_context *context= &lex->current_select->context;
             TABLE_LIST *save_list= context->table_list;
             const char *save_where= thd->where;
@@ -4181,12 +4318,12 @@ part_bit_expr:
               value_ptr->unsigned_flag= FALSE;
             if ((value_ptr->null_value= part_expr->null_value))
             {
-              if (Lex->part_info->curr_part_elem->has_null_value)
+              if (part_info->curr_part_elem->has_null_value)
               {
                 my_error(ER_MULTIPLE_DEF_CONST_IN_LIST_PART_ERROR, MYF(0));
                 MYSQL_YYABORT;
               }
-              Lex->part_info->curr_part_elem->has_null_value= TRUE;
+              part_info->curr_part_elem->has_null_value= TRUE;
             }
             else if (part_expr->result_type() != INT_RESULT)
             {
@@ -4200,8 +4337,9 @@ part_bit_expr:
 opt_sub_partition:
           /* empty */
           {
-            if (Lex->part_info->no_subparts != 0 &&
-                !Lex->part_info->use_default_subpartitions)
+            partition_info *part_info= Lex->part_info;
+            if (part_info->no_subparts != 0 &&
+                !part_info->use_default_subpartitions)
             {
               /*
                 We come here when we have defined subpartitions on the first
@@ -4213,8 +4351,7 @@ opt_sub_partition:
           }
         | '(' sub_part_list ')'
           {
-            LEX *lex= Lex;
-            partition_info *part_info= lex->part_info;
+            partition_info *part_info= Lex->part_info;
             if (part_info->no_subparts != 0)
             {
               if (part_info->no_subparts !=
@@ -4245,8 +4382,7 @@ sub_part_list:
 sub_part_definition:
           SUBPARTITION_SYM
           {
-            LEX *lex= Lex;
-            partition_info *part_info= lex->part_info;
+            partition_info *part_info= Lex->part_info;
             partition_element *curr_part= part_info->current_partition;
             partition_element *sub_p_elem= new partition_element(curr_part);
             if (part_info->use_default_subpartitions &&
@@ -4300,9 +4436,9 @@ opt_part_option:
           { Lex->part_info->curr_part_elem->tablespace_name= $3.str; }
         | opt_storage ENGINE_SYM opt_equal storage_engines
           {
-            LEX *lex= Lex;
-            lex->part_info->curr_part_elem->engine_type= $4;
-            lex->part_info->default_engine_type= $4;
+            partition_info *part_info= Lex->part_info;
+            part_info->curr_part_elem->engine_type= $4;
+            part_info->default_engine_type= $4;
           }
         | NODEGROUP_SYM opt_equal real_ulong_num
           { Lex->part_info->curr_part_elem->nodegroup_id= (uint16) $3; }
@@ -4318,6 +4454,11 @@ opt_part_option:
           { Lex->part_info->curr_part_elem->part_comment= $3.str; }
         ;
 
+opt_global:
+        /* empty */ { $$= FALSE;}
+        | GLOBAL_SYM { $$= TRUE; }
+        ;
+
 /*
  End of partition parser part
 */
@@ -5786,8 +5927,8 @@ reorg_parts_rule:
           }
           INTO '(' part_def_list ')'
           {
-            LEX *lex= Lex;
-            lex->part_info->no_parts= lex->part_info->partitions.elements;
+            partition_info *part_info= Lex->part_info;
+            part_info->no_parts= part_info->partitions.elements;
           }
         ;
 
@@ -11529,7 +11670,6 @@ keyword_sp:
         | MAX_SIZE_SYM             {}
         | MAX_UPDATES_PER_HOUR     {}
         | MAX_USER_CONNECTIONS_SYM {}
-        | MAX_VALUE_SYM            {}
         | MEDIUM_SYM               {}
         | MEMORY_SYM               {}
         | MERGE_SYM                {}