SQL: (0.6) Pruning for VERSIONING partitions [closes #97]

* based on RANGE pruning by COLUMNS (sys_trx_end) condition
* removed DEFAULT; AS OF NOW is always last; current VERSIONING as last non-empty (or first empty)
* Min/Max stats in TABLE_SHARE
* ALTER TABLE ADD PARTITION adds before AS OF NOW partition

14 files changed, 483 insertions, 138 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 1ba3210d62b..0fa461e1807 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4289,6 +4289,15 @@ int ha_partition::update_row(const uchar *old_data, uchar *new_data)
     if (error)
       goto exit;
 
+    if (m_part_info->part_type == VERSIONING_PARTITION)
+    {
+      uint sub_factor= m_part_info->num_subparts ? m_part_info->num_subparts : 1;
+      DBUG_ASSERT(m_tot_parts == m_part_info->num_parts * sub_factor);
+      uint lpart_id= new_part_id / sub_factor;
+      // lpart_id is VERSIONING partition because new_part_id != old_part_id
+      m_part_info->vers_update_stats(thd, lpart_id);
+    }
+
     tmp_disable_binlog(thd); /* Do not replicate the low-level changes. */
     error= m_file[old_part_id]->ha_delete_row(old_data);
     reenable_binlog(thd);
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 1ca320eef1a..2c7f4a0861f 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -1295,9 +1295,7 @@ public:
     {
       handler *file= m_file[part_id];
       DBUG_ASSERT(bitmap_is_set(&(m_part_info->read_partitions), part_id));
-      file->info(HA_STATUS_TIME | HA_STATUS_VARIABLE |
-        HA_STATUS_VARIABLE_EXTRA | HA_STATUS_NO_LOCK);
-
+      file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
       part_recs+= file->stats.records;
     }
     return part_recs;
diff --git a/sql/item.cc b/sql/item.cc
index bb1efca232b..410fa32f234 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -6989,6 +6989,26 @@ bool Item_temporal_literal::eq(const Item *item, bool binary_cmp) const
                      &((Item_temporal_literal *) item)->cached_time);
 }
 
+bool Item_temporal_literal::set_lower(MYSQL_TIME * ltime)
+{
+  if (my_time_compare(ltime, &cached_time) < 0)
+  {
+    cached_time= *ltime;
+    return true;
+  }
+  return false;
+}
+
+bool Item_temporal_literal::set_higher(MYSQL_TIME * ltime)
+{
+  if (my_time_compare(ltime, &cached_time) > 0)
+  {
+    cached_time= *ltime;
+    return true;
+  }
+  return false;
+}
+
 
 void Item_date_literal::print(String *str, enum_query_type query_type)
 {
diff --git a/sql/item.h b/sql/item.h
index b7381bde260..357eabd72e1 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -3865,6 +3865,12 @@ public:
   { return  val_decimal_from_date(decimal_value); }
   int save_in_field(Field *field, bool no_conversions)
   { return save_date_in_field(field, no_conversions); }
+  void set_time(MYSQL_TIME *ltime)
+  {
+    cached_time= *ltime;
+  }
+  bool set_lower(MYSQL_TIME *ltime);
+  bool set_higher(MYSQL_TIME *ltime);
 };
 
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index acda4b51627..5196ec41c11 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1017,7 +1017,8 @@ static PSI_mutex_info all_server_mutexes[]=
 
 PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,
   key_rwlock_LOCK_sys_init_connect, key_rwlock_LOCK_sys_init_slave,
-  key_rwlock_LOCK_system_variables_hash, key_rwlock_query_cache_query_lock;
+  key_rwlock_LOCK_system_variables_hash, key_rwlock_query_cache_query_lock,
+  key_rwlock_LOCK_vers_stats, key_rwlock_LOCK_stat_serial;
 
 static PSI_rwlock_info all_server_rwlocks[]=
 {
@@ -1029,7 +1030,9 @@ static PSI_rwlock_info all_server_rwlocks[]=
   { &key_rwlock_LOCK_sys_init_connect, "LOCK_sys_init_connect", PSI_FLAG_GLOBAL},
   { &key_rwlock_LOCK_sys_init_slave, "LOCK_sys_init_slave", PSI_FLAG_GLOBAL},
   { &key_rwlock_LOCK_system_variables_hash, "LOCK_system_variables_hash", PSI_FLAG_GLOBAL},
-  { &key_rwlock_query_cache_query_lock, "Query_cache_query::lock", 0}
+  { &key_rwlock_query_cache_query_lock, "Query_cache_query::lock", 0},
+  { &key_rwlock_LOCK_vers_stats, "Vers_field_stats::lock", 0},
+  { &key_rwlock_LOCK_stat_serial, "TABLE_SHARE::LOCK_stat_serial", 0}
 };
 
 #ifdef HAVE_MMAP
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 8ac0625b755..46382396a5a 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -314,7 +314,8 @@ extern PSI_mutex_key key_LOCK_gtid_waiting;
 
 extern PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,
   key_rwlock_LOCK_sys_init_connect, key_rwlock_LOCK_sys_init_slave,
-  key_rwlock_LOCK_system_variables_hash, key_rwlock_query_cache_query_lock;
+  key_rwlock_LOCK_system_variables_hash, key_rwlock_query_cache_query_lock,
+  key_rwlock_LOCK_vers_stats, key_rwlock_LOCK_stat_serial;
 
 #ifdef HAVE_MMAP
 extern PSI_cond_key key_PAGE_cond, key_COND_active, key_COND_pool;
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index fbdbf76ffd9..fcf038a212b 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -3454,6 +3454,11 @@ bool prune_partitions(THD *thd, TABLE *table, Item *pprune_cond)
     free_root(&alloc,MYF(0));		// Return memory & allocator
     DBUG_RETURN(FALSE);
   }
+
+  if (part_info->part_type == VERSIONING_PARTITION)
+  {
+    part_info->vers_update_range_constants(thd);
+  }
   
   dbug_tmp_use_all_columns(table, old_sets, 
                            table->read_set, table->write_set);
@@ -3980,7 +3985,7 @@ int find_used_partitions(PART_PRUNE_PARAM *ppar, SEL_ARG *key_tree)
         simply set res= -1 as if the mapper had returned that.
         TODO: What to do here is defined in WL#4065.
       */
-      if (ppar->arg_stack[0]->part == 0)
+      if (ppar->arg_stack[0]->part == 0 || ppar->part_info->part_type == VERSIONING_PARTITION)
       {
         uint32 i;
         uint32 store_length_array[MAX_KEY];
diff --git a/sql/partition_element.h b/sql/partition_element.h
index e76558bf7ae..18276ef713e 100644
--- a/sql/partition_element.h
+++ b/sql/partition_element.h
@@ -90,38 +90,64 @@ typedef struct p_elem_val
 
 struct st_ddl_log_memory_entry;
 
-class Stat_timestampf : public Sql_alloc
+class Vers_field_stats : public Sql_alloc
 {
   static const uint buf_size= 4 + (TIME_SECOND_PART_DIGITS + 1) / 2;
   uchar min_buf[buf_size];
   uchar max_buf[buf_size];
   Field_timestampf min_value;
   Field_timestampf max_value;
+  mysql_rwlock_t lock;
 
 public:
-  Stat_timestampf(const char *field_name, TABLE_SHARE *share) :
+  Vers_field_stats(const char *field_name, TABLE_SHARE *share) :
     min_value(min_buf, NULL, 0, Field::NONE, field_name, share, 6),
     max_value(max_buf, NULL, 0, Field::NONE, field_name, share, 6)
   {
     min_value.set_max();
     memset(max_buf, 0, buf_size);
+    mysql_rwlock_init(key_rwlock_LOCK_vers_stats, &lock);
   }
-  void update(Field *from)
+  ~Vers_field_stats()
   {
-    from->update_min(&min_value, false);
-    from->update_max(&max_value, false);
+    mysql_rwlock_destroy(&lock);
+  }
+  bool update_unguarded(Field *from)
+  {
+    return
+      from->update_min(&min_value, false) +
+      from->update_max(&max_value, false);
+  }
+  bool update(Field *from)
+  {
+    mysql_rwlock_wrlock(&lock);
+    bool res= update_unguarded(from);
+    mysql_rwlock_unlock(&lock);
+    return res;
   }
   my_time_t min_time()
   {
-    return min_value.get_timestamp();
+    mysql_rwlock_rdlock(&lock);
+    my_time_t res= min_value.get_timestamp();
+    mysql_rwlock_unlock(&lock);
+    return res;
   }
   my_time_t max_time()
   {
-    return max_value.get_timestamp();
+    mysql_rwlock_rdlock(&lock);
+    my_time_t res= max_value.get_timestamp();
+    mysql_rwlock_unlock(&lock);
+    return res;
   }
 };
 
-class partition_element :public Sql_alloc {
+enum stat_trx_field
+{
+  STAT_TRX_END= 0
+};
+
+class partition_element :public Sql_alloc
+{
 public:
   List<partition_element> subpartitions;
   List<part_elem_value> list_val_list;
@@ -142,6 +168,7 @@ public:
   bool signed_flag;                          // Range value signed
   bool max_value;                            // MAXVALUE range
   uint32 id;
+  bool empty;
 
   enum elem_type
   {
@@ -151,7 +178,6 @@ public:
   };
 
   elem_type type;
-  Stat_timestampf *stat_trx_end;
 
   partition_element()
   : part_max_rows(0), part_min_rows(0), range_value(0),
@@ -162,10 +188,9 @@ public:
     nodegroup_id(UNDEF_NODEGROUP), has_null_value(FALSE),
     signed_flag(FALSE), max_value(FALSE),
     id(UINT32_MAX),
-    type(CONVENTIONAL),
-    stat_trx_end(NULL)
-  {
-  }
+    empty(true),
+    type(CONVENTIONAL)
+  {}
   partition_element(partition_element *part_elem)
   : part_max_rows(part_elem->part_max_rows),
     part_min_rows(part_elem->part_min_rows),
@@ -180,11 +205,19 @@ public:
     nodegroup_id(part_elem->nodegroup_id),
     has_null_value(FALSE),
     id(part_elem->id),
-    type(part_elem->type),
-    stat_trx_end(NULL)
+    empty(part_elem->empty),
+    type(part_elem->type)
+  {}
+  ~partition_element() {}
+
+  part_column_list_val& get_col_val(uint idx)
   {
+    DBUG_ASSERT(type != CONVENTIONAL);
+    DBUG_ASSERT(list_val_list.elements == 1);
+    part_elem_value *ev= static_cast<part_elem_value*>(list_val_list.first_node()->info);
+    DBUG_ASSERT(ev && ev->col_val_array);
+    return ev->col_val_array[idx];
   }
-  ~partition_element() {}
 };
 
 #endif /* PARTITION_ELEMENT_INCLUDED */
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index e83fced4495..f45b45548b0 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -21,6 +21,7 @@
 #endif
 
 #include <my_global.h>
+#include <tztime.h>
 #include "sql_priv.h"
 // Required to get server definitions for mysql/plugin.h right
 #include "sql_plugin.h"
@@ -796,6 +797,7 @@ bool partition_info::vers_init_info(THD * thd)
   part_type= VERSIONING_PARTITION;
   list_of_part_fields= TRUE;
   column_list= TRUE;
+  num_columns= 1;
   vers_info= new (thd->mem_root) Vers_part_info;
   if (!vers_info)
   {
@@ -854,8 +856,11 @@ partition_element*
 partition_info::vers_part_rotate(THD * thd)
 {
   DBUG_ASSERT(table && table->s);
-  if (table->s->free_parts.is_empty())
+  DBUG_ASSERT(vers_info && vers_info->initialized());
+
+  if (table->s->hist_part_id >= vers_info->now_part->id - 1)
   {
+    DBUG_ASSERT(table->s->hist_part_id == vers_info->now_part->id - 1);
     push_warning_printf(thd,
       Sql_condition::WARN_LEVEL_WARN,
       WARN_VERS_PART_FULL,
@@ -864,8 +869,7 @@ partition_info::vers_part_rotate(THD * thd)
     return vers_info->hist_part;
   }
 
-  table->s->vers_part_rotate();
-  DBUG_ASSERT(table->s->hist_part_id < num_parts);
+  table->s->hist_part_id++;
   const char* old_part_name= vers_info->hist_part->partition_name;
   vers_hist_part();
 
@@ -879,18 +883,100 @@ partition_info::vers_part_rotate(THD * thd)
   return vers_info->hist_part;
 }
 
-bool partition_info::vers_setup_1(THD * thd)
+bool partition_info::vers_setup_1(THD * thd, uint32 added)
 {
   DBUG_ASSERT(part_type == VERSIONING_PARTITION);
+
   if (!table->versioned())
   {
     my_error(ER_VERSIONING_REQUIRED, MYF(0), "`BY SYSTEM_TIME` partitioning");
     return true;
   }
-  Field *sys_trx_end= table->vers_end_field();
-  part_field_list.empty();
-  part_field_list.push_back(const_cast<char *>(sys_trx_end->field_name), thd->mem_root);
-  sys_trx_end->flags|= GET_FIXED_FIELDS_FLAG; // needed in handle_list_of_fields()
+
+  if (added)
+  {
+    DBUG_ASSERT(partitions.elements > added + 1);
+    Vers_field_stats** old_array= table->s->stat_trx;
+    table->s->stat_trx= static_cast<Vers_field_stats**>(
+      alloc_root(&table->s->mem_root, sizeof(void *) * partitions.elements * num_columns));
+    memcpy(table->s->stat_trx, old_array, sizeof(void *) * (partitions.elements - added) * num_columns);
+  }
+  else
+  {
+    Field *sys_trx_end= table->vers_end_field();
+    part_field_list.empty();
+    part_field_list.push_back(const_cast<char *>(sys_trx_end->field_name), thd->mem_root);
+    DBUG_ASSERT(part_field_list.elements == num_columns);
+    // needed in handle_list_of_fields()
+    sys_trx_end->flags|= GET_FIXED_FIELDS_FLAG;
+  }
+
+  List_iterator<partition_element> it(partitions);
+  partition_element *el;
+  MYSQL_TIME t;
+  memset(&t, 0, sizeof(t));
+  my_time_t ts= TIMESTAMP_MAX_VALUE - partitions.elements;
+  uint32 id= 0;
+  while ((el= it++))
+  {
+    DBUG_ASSERT(el->type != partition_element::CONVENTIONAL);
+    ++ts;
+    if (added)
+    {
+      if (el->type == partition_element::VERSIONING && !el->empty)
+      {
+        ++id;
+        continue;
+      }
+      if (el->id == UINT32_MAX || el->type == partition_element::AS_OF_NOW)
+      {
+        DBUG_ASSERT(table && table->s);
+        Vers_field_stats *stat_trx_end= new (&table->s->mem_root)
+          Vers_field_stats(table->s->vers_end_field()->field_name, table->s);
+        table->s->stat_trx[id * num_columns + STAT_TRX_END]= stat_trx_end;
+        el->id= id++;
+        if (el->type == partition_element::AS_OF_NOW)
+          break;
+        goto create_col_val;
+      }
+      thd->variables.time_zone->gmt_sec_to_TIME(&t, ts);
+      for (uint i= 0; i < num_columns; ++i)
+      {
+        part_column_list_val &col_val= el->get_col_val(i);
+        static_cast<Item_datetime_literal *>(col_val.item_expression)->set_time(&t);
+        col_val.fixed= 0;
+      }
+      ++id;
+      continue;
+    }
+
+  create_col_val:
+    curr_part_elem= el;
+    init_column_part(thd);
+    el->list_val_list.empty();
+    el->list_val_list.push_back(curr_list_val, thd->mem_root);
+    thd->variables.time_zone->gmt_sec_to_TIME(&t, ts);
+    for (uint i= 0; i < num_columns; ++i)
+    {
+      part_column_list_val *col_val= add_column_value(thd);
+      if (el->type == partition_element::AS_OF_NOW)
+      {
+        col_val->max_value= true;
+        col_val->item_expression= NULL;
+        col_val->column_value= NULL;
+        col_val->part_info= this;
+        col_val->fixed= 1;
+        continue;
+      }
+      Item *item_expression= new (thd->mem_root) Item_datetime_literal(thd, &t);
+      /* We initialize col_val with bogus max value to make fix_partition_func() and check_range_constants() happy.
+         Later in vers_setup_2() it is initialized with real stat value if there will be any. */
+      /* FIXME: TIME_RESULT in col_val is expensive. It should be INT_RESULT
+        (got to be fixed when InnoDB is supported). */
+      init_col_val(col_val, item_expression);
+      DBUG_ASSERT(item_expression == el->get_col_val(i).item_expression);
+    }
+  }
   return false;
 }
 
@@ -902,6 +988,8 @@ bool partition_info::vers_scan_min_max(THD *thd, partition_element *part)
   uint32 sub_factor= num_subparts ? num_subparts : 1;
   uint32 part_id= part->id * sub_factor;
   uint32 part_id_end= part_id + sub_factor;
+  DBUG_ASSERT(part->empty);
+  DBUG_ASSERT(table->s->stat_trx);
   for (; part_id < part_id_end; ++part_id)
   {
     handler *file= table->file->part_handler(part_id);
@@ -913,6 +1001,8 @@ bool partition_info::vers_scan_min_max(THD *thd, partition_element *part)
     {
       while ((rc= file->ha_rnd_next(table->record[0])) != HA_ERR_END_OF_FILE)
       {
+        if (part->empty)
+          part->empty= false;
         if (thd->killed)
         {
           file->ha_rnd_end();
@@ -924,7 +1014,7 @@ bool partition_info::vers_scan_min_max(THD *thd, partition_element *part)
             continue;
           break;
         }
-        part->stat_trx_end->update(table->vers_end_field());
+        vers_stat_trx(STAT_TRX_END, part).update_unguarded(table->vers_end_field());
       }
       file->ha_rnd_end();
     }
@@ -939,12 +1029,49 @@ bool partition_info::vers_scan_min_max(THD *thd, partition_element *part)
   return false;
 }
 
+void partition_info::vers_update_col_vals(THD *thd, partition_element *el0, partition_element *el1)
+{
+  MYSQL_TIME t;
+  memset(&t, 0, sizeof(t));
+  DBUG_ASSERT(table && table->s && table->s->stat_trx);
+  DBUG_ASSERT(!el0 || el1->id == el0->id + 1);
+  const uint idx= el1->id * num_columns;
+  my_time_t ts;
+  part_column_list_val *col_val;
+  Item_datetime_literal *val_item;
+  Vers_field_stats *stat_trx_x;
+  for (uint i= 0; i < num_columns; ++i)
+  {
+    stat_trx_x= table->s->stat_trx[idx + i];
+    if (el0)
+    {
+      ts= stat_trx_x->min_time();
+      thd->variables.time_zone->gmt_sec_to_TIME(&t, ts);
+      col_val= &el0->get_col_val(i);
+      val_item= static_cast<Item_datetime_literal*>(col_val->item_expression);
+      DBUG_ASSERT(val_item);
+      if (val_item->set_lower(&t))
+        col_val->fixed= 0;
+    }
+    col_val= &el1->get_col_val(i);
+    if (!col_val->max_value)
+    {
+      ts= stat_trx_x->max_time() + 1;
+      thd->variables.time_zone->gmt_sec_to_TIME(&t, ts);
+      val_item= static_cast<Item_datetime_literal*>(col_val->item_expression);
+      DBUG_ASSERT(val_item);
+      if (val_item->set_higher(&t))
+        col_val->fixed= 0;
+    }
+  }
+}
+
 
 // setup at open stage (TABLE_SHARE is initialized)
 bool partition_info::vers_setup_2(THD * thd, bool is_create_table_ind)
 {
   DBUG_ASSERT(part_type == VERSIONING_PARTITION);
-  DBUG_ASSERT(vers_info && vers_info->initialized(is_create_table_ind) && vers_info->hist_default != UINT32_MAX);
+  DBUG_ASSERT(vers_info && vers_info->initialized(false));
   DBUG_ASSERT(table && table->s);
   if (!table->versioned_by_sql())
   {
@@ -961,38 +1088,78 @@ bool partition_info::vers_setup_2(THD * thd, bool is_create_table_ind)
   {
     table->s->busy_rotation= true;
     mysql_mutex_unlock(&table->s->LOCK_rotation);
+
+    DBUG_ASSERT(part_field_list.elements == num_columns);
+
+    bool dont_stat= true;
+    bool col_val_updated= false;
+    if (!table->s->stat_trx)
+    {
+      DBUG_ASSERT(partitions.elements > 1);
+      table->s->stat_trx= static_cast<Vers_field_stats**>(
+        alloc_root(&table->s->mem_root, sizeof(void *) * partitions.elements * num_columns));
+      dont_stat= false;
+    }
+
     // build freelist, scan min/max, assign hist_part
     List_iterator<partition_element> it(partitions);
-    partition_element *el;
-    while ((el= it++))
+    partition_element *el= NULL, *prev;
+    while ((prev= el, el= it++))
     {
-      DBUG_ASSERT(el->type != partition_element::CONVENTIONAL);
-      if (el->type == partition_element::VERSIONING)
+      if (el->type == partition_element::VERSIONING && dont_stat)
+      {
+        if (el->id == table->s->hist_part_id)
+        {
+          vers_info->hist_part= el;
+          break;
+        }
+        continue;
+      }
+
+      {
+        Vers_field_stats *stat_trx_end= new (&table->s->mem_root)
+          Vers_field_stats(table->s->vers_end_field()->field_name, table->s);
+        table->s->stat_trx[el->id * num_columns + STAT_TRX_END]= stat_trx_end;
+      }
+
+      if (!is_create_table_ind)
       {
-        DBUG_ASSERT(!el->stat_trx_end);
-        el->stat_trx_end= new (&table->mem_root)
-          Stat_timestampf(table->s->vers_end_field()->field_name, table->s);
-        if (!is_create_table_ind && vers_scan_min_max(thd, el))
+        if (vers_scan_min_max(thd, el))
           return true;
+        if (!el->empty)
+        {
+          vers_update_col_vals(thd, prev, el);
+          col_val_updated= true;
+        }
       }
-      if (el == vers_info->now_part || el == vers_info->hist_part)
-        continue;
-      if (!vers_info->hist_part && el->id == vers_info->hist_default)
+
+      if (el->type == partition_element::AS_OF_NOW)
+        break;
+
+      DBUG_ASSERT(el->type == partition_element::VERSIONING);
+
+      if (vers_info->hist_part)
       {
-        vers_info->hist_part= el;
+        if (!el->empty)
+          goto set_hist_part;
       }
-      if (is_create_table_ind || (
-        table->s->free_parts_init &&
-        !vers_limit_exceed(el) &&
-        !vers_interval_exceed(el)))
+      else
       {
-        table->s->free_parts.push_back((void *) el->id, &table->s->mem_root);
+      set_hist_part:
+        vers_info->hist_part= el;
+        continue;
       }
+    } // while
+
+    if (!dont_stat)
+    {
+      if (col_val_updated)
+        table->s->stat_serial++;
+
+      table->s->hist_part_id= vers_info->hist_part->id;
+      if (!is_create_table_ind && (vers_limit_exceed() || vers_interval_exceed()))
+        vers_part_rotate(thd);
     }
-    table->s->hist_part_id= vers_info->hist_part->id;
-    if (!is_create_table_ind && (vers_limit_exceed() || vers_interval_exceed()))
-      vers_part_rotate(thd);
-    table->s->free_parts_init= false;
     mysql_mutex_lock(&table->s->LOCK_rotation);
     mysql_cond_broadcast(&table->s->COND_rotation);
     table->s->busy_rotation= false;
@@ -1184,7 +1351,7 @@ error:
     called for RANGE PARTITIONed tables.
 */
 
-bool partition_info::check_range_constants(THD *thd)
+bool partition_info::check_range_constants(THD *thd, bool init)
 {
   partition_element* part_def;
   bool first= TRUE;
@@ -1201,12 +1368,15 @@ bool partition_info::check_range_constants(THD *thd)
     part_column_list_val *UNINIT_VAR(current_largest_col_val);
     uint num_column_values= part_field_list.elements;
     uint size_entries= sizeof(part_column_list_val) * num_column_values;
-    range_col_array= (part_column_list_val*) thd->calloc(num_parts *
-                                                         size_entries);
-    if (unlikely(range_col_array == NULL))
+    if (init)
     {
-      mem_alloc_error(num_parts * size_entries);
-      goto end;
+      range_col_array= (part_column_list_val*) thd->calloc(num_parts *
+        size_entries);
+      if (unlikely(range_col_array == NULL))
+      {
+        mem_alloc_error(num_parts * size_entries);
+        goto end;
+      }
     }
     loc_range_col_array= range_col_array;
     i= 0;
@@ -1239,11 +1409,14 @@ bool partition_info::check_range_constants(THD *thd)
     longlong part_range_value;
     bool signed_flag= !part_expr->unsigned_flag;
 
-    range_int_array= (longlong*) thd->alloc(num_parts * sizeof(longlong));
-    if (unlikely(range_int_array == NULL))
+    if (init)
     {
-      mem_alloc_error(num_parts * sizeof(longlong));
-      goto end;
+      range_int_array= (longlong*) thd->alloc(num_parts * sizeof(longlong));
+      if (unlikely(range_int_array == NULL))
+      {
+        mem_alloc_error(num_parts * sizeof(longlong));
+        goto end;
+      }
     }
     i= 0;
     do
@@ -1609,7 +1782,6 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
   char *same_name;
   uint32 hist_parts= 0;
   uint32 now_parts= 0;
-  const char* hist_default= NULL;
   DBUG_ENTER("partition_info::check_partition_info");
   DBUG_ASSERT(default_engine_type != partition_hton);
 
@@ -1816,13 +1988,6 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
         if (part_elem->type == partition_element::VERSIONING)
         {
           hist_parts++;
-          if (vers_info->hist_default == UINT32_MAX)
-          {
-            vers_info->hist_default= part_elem->id;
-            hist_default= part_elem->partition_name;
-          }
-          if (vers_info->hist_default == part_elem->id)
-            vers_info->hist_part= part_elem;
         }
         else
         {
@@ -1861,7 +2026,7 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
 
   if (add_or_reorg_part)
   {
-    if (unlikely((part_type == RANGE_PARTITION &&
+    if (unlikely(((part_type == RANGE_PARTITION || part_type == VERSIONING_PARTITION) &&
                   check_range_constants(thd)) ||
                  (part_type == LIST_PARTITION &&
                   check_list_constants(thd))))
@@ -1878,14 +2043,6 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
         ER_THD(thd, WARN_VERS_PARAMETERS),
         "no rotation condition for multiple `VERSIONING` partitions.");
     }
-    if (hist_default)
-    {
-      push_warning_printf(thd,
-        Sql_condition::WARN_LEVEL_WARN,
-        WARN_VERS_PARAMETERS,
-        "No `DEFAULT` for `VERSIONING` partitions. Setting `%s` as default.",
-        hist_default);
-    }
   }
   if (now_parts > 1)
   {
diff --git a/sql/partition_info.h b/sql/partition_info.h
index 844956594ff..5a671bfc50f 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -41,7 +41,7 @@ struct Vers_part_info : public Sql_alloc
     limit(0),
     now_part(NULL),
     hist_part(NULL),
-    hist_default(UINT32_MAX)
+    stat_serial(0)
   {
   }
   Vers_part_info(Vers_part_info &src) :
@@ -49,7 +49,7 @@ struct Vers_part_info : public Sql_alloc
     limit(src.limit),
     now_part(NULL),
     hist_part(NULL),
-    hist_default(src.hist_default)
+    stat_serial(src.stat_serial)
   {
   }
   bool initialized(bool fully= true)
@@ -71,7 +71,7 @@ struct Vers_part_info : public Sql_alloc
   ulonglong limit;
   partition_element *now_part;
   partition_element *hist_part;
-  uint32 hist_default;
+  ulonglong stat_serial;
 };
 
 class partition_info : public Sql_alloc
@@ -182,8 +182,9 @@ public:
     LIST_PART_ENTRY *list_array;
     part_column_list_val *range_col_array;
     part_column_list_val *list_col_array;
-    Vers_part_info *vers_info;
   };
+
+  Vers_part_info *vers_info;
   
   /********************************************
    * INTERVAL ANALYSIS
@@ -350,7 +351,7 @@ public:
   char *find_duplicate_field();
   char *find_duplicate_name();
   bool check_engine_mix(handlerton *engine_type, bool default_engine);
-  bool check_range_constants(THD *thd);
+  bool check_range_constants(THD *thd, bool init= true);
   bool check_list_constants(THD *thd);
   bool check_partition_info(THD *thd, handlerton **eng_type,
                             handler *file, HA_CREATE_INFO *info,
@@ -400,9 +401,10 @@ public:
   bool vers_set_interval(const INTERVAL &i);
   bool vers_set_limit(ulonglong limit);
   partition_element* vers_part_rotate(THD *thd);
-  bool vers_setup_1(THD *thd);
+  bool vers_setup_1(THD *thd, uint32 added= 0);
   bool vers_setup_2(THD *thd, bool is_create_table_ind);
   bool vers_scan_min_max(THD *thd, partition_element *part);
+  void vers_update_col_vals(THD *thd, partition_element *el0, partition_element *el1);
 
   partition_element *vers_hist_part()
   {
@@ -427,6 +429,17 @@ public:
     DBUG_ASSERT(0);
     return NULL;
   }
+  partition_element *get_partition(uint part_id)
+  {
+    List_iterator<partition_element> it(partitions);
+    partition_element *el;
+    while ((el= it++))
+    {
+      if (el->id == part_id)
+        return el;
+    }
+    return NULL;
+  }
   bool vers_limit_exceed(partition_element *part= NULL)
   {
     DBUG_ASSERT(vers_info);
@@ -440,6 +453,18 @@ public:
     // TODO: cache thread-shared part_recs and increment on INSERT
     return table->file->part_recs_slow(part) >= vers_info->limit;
   }
+  Vers_field_stats& vers_stat_trx(stat_trx_field fld, uint32 part_element_id)
+  {
+    DBUG_ASSERT(table && table->s && table->s->stat_trx);
+    Vers_field_stats* res= table->s->stat_trx[part_element_id * num_columns + fld];
+    DBUG_ASSERT(res);
+    return *res;
+  }
+  Vers_field_stats& vers_stat_trx(stat_trx_field fld, partition_element *part)
+  {
+    DBUG_ASSERT(part);
+    return vers_stat_trx(fld, part->id);
+  }
   bool vers_interval_exceed(my_time_t max_time, partition_element *part= NULL)
   {
     DBUG_ASSERT(vers_info);
@@ -450,19 +475,63 @@ public:
       DBUG_ASSERT(vers_info->initialized());
       part= vers_hist_part();
     }
-    DBUG_ASSERT(part->stat_trx_end);
-    max_time-= part->stat_trx_end->min_time();
+    max_time-= vers_stat_trx(STAT_TRX_END, part).min_time();
     return max_time > vers_info->interval;
   }
   bool vers_interval_exceed(partition_element *part)
   {
-    DBUG_ASSERT(part->stat_trx_end);
-    return vers_interval_exceed(part->stat_trx_end->max_time(), part);
+    return vers_interval_exceed(vers_stat_trx(STAT_TRX_END, part).max_time(), part);
   }
   bool vers_interval_exceed()
   {
     return vers_interval_exceed(vers_hist_part());
   }
+  void vers_update_stats(THD *thd, partition_element *el)
+  {
+    DBUG_ASSERT(vers_info && vers_info->initialized());
+    DBUG_ASSERT(table && table->s);
+    DBUG_ASSERT(el && el->type == partition_element::VERSIONING);
+    mysql_rwlock_wrlock(&table->s->LOCK_stat_serial);
+    el->empty= false;
+    bool updated=
+      vers_stat_trx(STAT_TRX_END, el->id).update(table->vers_end_field());
+    if (updated)
+      table->s->stat_serial++;
+    mysql_rwlock_unlock(&table->s->LOCK_stat_serial);
+    if (updated)
+    {
+      vers_update_col_vals(thd,
+        el->id > 0 ? get_partition(el->id - 1) : NULL,
+        el);
+    }
+  }
+  void vers_update_stats(THD *thd, uint part_id)
+  {
+    DBUG_ASSERT(vers_info && vers_info->initialized());
+    if (part_id < vers_info->now_part->id)
+      vers_update_stats(thd, get_partition(part_id));
+  }
+  void vers_update_range_constants(THD *thd)
+  {
+    DBUG_ASSERT(vers_info && vers_info->initialized());
+    DBUG_ASSERT(table && table->s);
+
+    mysql_rwlock_rdlock(&table->s->LOCK_stat_serial);
+    if (vers_info->stat_serial == table->s->stat_serial)
+    {
+      mysql_rwlock_unlock(&table->s->LOCK_stat_serial);
+      return;
+    }
+
+    for (uint i= 0; i < num_columns; ++i)
+    {
+      Field *f= part_field_array[i];
+      bitmap_set_bit(f->table->write_set, f->field_index);
+    }
+    check_range_constants(thd, false);
+    vers_info->stat_serial= table->s->stat_serial;
+    mysql_rwlock_unlock(&table->s->LOCK_stat_serial);
+  }
 };
 
 uint32 get_next_partition_id_range(struct st_partition_iter* part_iter);
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 816cd374e67..b358fe3386e 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -1716,6 +1716,8 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else if (part_info->part_type == VERSIONING_PARTITION)
     {
       error_str= partition_keywords[PKW_SYSTEM_TIME].str;
+      if (unlikely(part_info->check_range_constants(thd)))
+        goto end;
     }
     else
     {
@@ -2396,9 +2398,6 @@ static int add_partition_values(File fptr, partition_info *part_info,
       break;
     case partition_element::VERSIONING:
       err+= add_string(fptr, " VERSIONING");
-      DBUG_ASSERT(part_info->vers_info);
-      if (part_info->vers_info->hist_default == p_elem->id)
-        err+= add_string(fptr, " DEFAULT");
       break;
     default:
       DBUG_ASSERT(0 && "wrong p_elem->type");
@@ -3423,7 +3422,8 @@ int vers_get_partition_id(partition_info *part_info,
                           longlong *func_value)
 {
   DBUG_ENTER("vers_get_partition_id");
-  Field *sys_trx_end= part_info->part_field_array[0];
+  DBUG_ASSERT(part_info);
+  Field *sys_trx_end= part_info->part_field_array[STAT_TRX_END];
   DBUG_ASSERT(sys_trx_end);
   DBUG_ASSERT(part_info->table);
   Vers_part_info *vers_info= part_info->vers_info;
@@ -3438,7 +3438,6 @@ int vers_get_partition_id(partition_info *part_info,
   }
   else // row is historical
   {
-    partition_element *part= vers_info->hist_part;
     THD *thd= current_thd;
     TABLE *table= part_info->table;
 
@@ -3461,18 +3460,13 @@ int vers_get_partition_id(partition_info *part_info,
         mysql_mutex_unlock(&table->s->LOCK_rotation);
         if (part_info->vers_limit_exceed() || part_info->vers_interval_exceed(sys_trx_end->get_timestamp()))
         {
-          part= part_info->vers_part_rotate(thd);
+          part_info->vers_part_rotate(thd);
         }
         mysql_mutex_lock(&table->s->LOCK_rotation);
         mysql_cond_broadcast(&table->s->COND_rotation);
         table->s->busy_rotation= false;
       }
       mysql_mutex_unlock(&table->s->LOCK_rotation);
-      if (vers_info->interval)
-      {
-        DBUG_ASSERT(part->stat_trx_end);
-        part->stat_trx_end->update(sys_trx_end);
-      }
       break;
     default:
       ;
@@ -5295,6 +5289,21 @@ that are reorganised.
         partition configuration is made.
       */
       {
+        partition_element *now_part= NULL;
+        if (tab_part_info->part_type == VERSIONING_PARTITION)
+        {
+          List_iterator<partition_element> it(tab_part_info->partitions);
+          partition_element *el;
+          while ((el= it++))
+          {
+            if (el->type == partition_element::AS_OF_NOW)
+            {
+              DBUG_ASSERT(tab_part_info->vers_info && el == tab_part_info->vers_info->now_part);
+              it.remove();
+              now_part= el;
+            }
+          }
+        }
         List_iterator<partition_element> alt_it(alt_part_info->partitions);
         uint part_count= 0;
         do
@@ -5309,6 +5318,15 @@ that are reorganised.
           }
         } while (++part_count < num_new_partitions);
         tab_part_info->num_parts+= num_new_partitions;
+        if (tab_part_info->part_type == VERSIONING_PARTITION)
+        {
+          DBUG_ASSERT(now_part);
+          if (tab_part_info->partitions.push_back(now_part, thd->mem_root))
+          {
+            mem_alloc_error(1);
+            goto err;
+          }
+        }
       }
       /*
         If we specify partitions explicitly we don't use defaults anymore.
@@ -5697,6 +5715,12 @@ the generated partition syntax in a correct manner.
         tab_part_info->use_default_subpartitions= FALSE;
         tab_part_info->use_default_num_subpartitions= FALSE;
       }
+
+      if (alter_info->flags & Alter_info::ALTER_ADD_PARTITION &&
+        tab_part_info->part_type == VERSIONING_PARTITION &&
+        tab_part_info->vers_setup_1(thd, alt_part_info->partitions.elements))
+        goto err;
+
       if (tab_part_info->check_partition_info(thd, (handlerton**)NULL,
                                               table->file, 0, TRUE))
       {
@@ -7548,6 +7572,7 @@ static void set_up_range_analysis_info(partition_info *part_info)
   switch (part_info->part_type) {
   case RANGE_PARTITION:
   case LIST_PARTITION:
+  case VERSIONING_PARTITION:
     if (!part_info->column_list)
     {
       if (part_info->part_expr->get_monotonicity_info() != NON_MONOTONIC)
@@ -7848,7 +7873,7 @@ int get_part_iter_for_interval_cols_via_map(partition_info *part_info,
   uint full_length= 0;
   DBUG_ENTER("get_part_iter_for_interval_cols_via_map");
 
-  if (part_info->part_type == RANGE_PARTITION)
+  if (part_info->part_type == RANGE_PARTITION || part_info->part_type == VERSIONING_PARTITION)
   {
     get_col_endpoint= get_partition_id_cols_range_for_endpoint;
     part_iter->get_next= get_next_partition_id_range;
@@ -7894,7 +7919,7 @@ int get_part_iter_for_interval_cols_via_map(partition_info *part_info,
   }
   if (flags & NO_MAX_RANGE)
   {
-    if (part_info->part_type == RANGE_PARTITION)
+    if (part_info->part_type == RANGE_PARTITION || part_info->part_type == VERSIONING_PARTITION)
       part_iter->part_nums.end= part_info->num_parts;
     else /* LIST_PARTITION */
     {
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 2f1346eeb40..f66c79a52ca 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -5197,6 +5197,7 @@ opt_part_values:
           {
             LEX *lex= Lex;
             partition_info *part_info= lex->part_info;
+            partition_element *elem= part_info->curr_part_elem;
             if (! lex->is_partition_management())
             {
               if (part_info->part_type != VERSIONING_PARTITION)
@@ -5205,17 +5206,22 @@ opt_part_values:
             }
             else
             {
-              part_info->vers_init_info(thd);
+              // FIXME: other ALTER commands?
+              my_yyabort_error((ER_VERS_WRONG_PARAMS, MYF(0), "BY SYSTEM_TIME", "AS OF NOW partition can not be added"));
             }
-            partition_element *elem= part_info->curr_part_elem;
             elem->type= partition_element::AS_OF_NOW;
             DBUG_ASSERT(part_info->vers_info);
             part_info->vers_info->now_part= elem;
+            if (part_info->init_column_part(thd))
+            {
+              MYSQL_YYABORT;
+            }
           }
         | VERSIONING_SYM
           {
             LEX *lex= Lex;
             partition_info *part_info= lex->part_info;
+            partition_element *elem= part_info->curr_part_elem;
             if (! lex->is_partition_management())
             {
               if (part_info->part_type != VERSIONING_PARTITION)
@@ -5225,10 +5231,17 @@ opt_part_values:
             else
             {
               part_info->vers_init_info(thd);
+              elem->id= UINT32_MAX;
+            }
+            DBUG_ASSERT(part_info->vers_info);
+            if (part_info->vers_info->now_part)
+              my_yyabort_error((ER_VERS_WRONG_PARAMS, MYF(0), "BY SYSTEM_TIME", "AS OF NOW partition is not last"));
+            elem->type= partition_element::VERSIONING;
+            if (part_info->init_column_part(thd))
+            {
+              MYSQL_YYABORT;
             }
-            part_info->curr_part_elem->type= partition_element::VERSIONING;
           }
-          opt_default_hist_part
         | DEFAULT
          {
             LEX *lex= Lex;
@@ -5556,20 +5569,6 @@ opt_versioning_limit:
          }
        ;
 
-opt_default_hist_part:
-          /* empty */ {}
-        | DEFAULT
-          {
-            partition_info *part_info= Lex->part_info;
-            DBUG_ASSERT(part_info && part_info->vers_info && part_info->curr_part_elem);
-            if (part_info->vers_info->hist_part)
-              my_yyabort_error((ER_VERS_WRONG_PARAMS, MYF(0),
-                "BY SYSTEM_TIME", "multiple `DEFAULT` partitions"));
-            part_info->vers_info->hist_part= part_info->curr_part_elem;
-            part_info->vers_info->hist_default= part_info->curr_part_elem->id;
-          }
-        ;
-
 /*
  End of partition parser part
 */
diff --git a/sql/table.cc b/sql/table.cc
index 4b7d5ef0d6f..354658ba476 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -3226,6 +3226,7 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
   }
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
+  bool work_part_info_used;
   if (share->partition_info_str_len && outparam->file)
   {
   /*
@@ -3246,7 +3247,6 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
     thd->set_n_backup_active_arena(&part_func_arena, &backup_arena);
     thd->stmt_arena= &part_func_arena;
     bool tmp;
-    bool work_part_info_used;
 
     tmp= mysql_unpack_partition(thd, share->partition_info_str,
                                 share->partition_info_str_len,
@@ -3412,12 +3412,32 @@ partititon_err:
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   if (outparam->part_info &&
-    outparam->part_info->part_type == VERSIONING_PARTITION &&
-    outparam->part_info->vers_setup_2(thd, is_create_table))
+    outparam->part_info->part_type == VERSIONING_PARTITION)
   {
-    error= OPEN_FRM_OPEN_ERROR;
-    error_reported= true;
-    goto err;
+    Query_arena *backup_stmt_arena_ptr= thd->stmt_arena;
+    Query_arena backup_arena;
+    Query_arena part_func_arena(&outparam->mem_root,
+                                Query_arena::STMT_INITIALIZED);
+    if (!work_part_info_used)
+    {
+      thd->set_n_backup_active_arena(&part_func_arena, &backup_arena);
+      thd->stmt_arena= &part_func_arena;
+    }
+
+    bool err= outparam->part_info->vers_setup_2(thd, is_create_table);
+
+    if (!work_part_info_used)
+    {
+      thd->stmt_arena= backup_stmt_arena_ptr;
+      thd->restore_active_arena(&part_func_arena, &backup_arena);
+    }
+
+    if (err)
+    {
+      error= OPEN_FRM_OPEN_ERROR;
+      error_reported= true;
+      goto err;
+    }
   }
 #endif
 
diff --git a/sql/table.h b/sql/table.h
index cb24d211261..402b2548873 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -561,6 +561,8 @@ struct TABLE_STATISTICS_CB
   bool histograms_are_read;   
 };
 
+class Vers_field_stats;
+
 #ifndef UINT32_MAX
 #define UINT32_MAX             (4294967295U)
 #endif
@@ -752,26 +754,30 @@ struct TABLE_SHARE
   uint16 row_start_field;
   uint16 row_end_field;
   uint32 hist_part_id;
-  List<void> free_parts;
-  bool free_parts_init;
+  Vers_field_stats** stat_trx;
+  ulonglong stat_serial; // guards check_range_constants() updates
+
   bool busy_rotation;
   mysql_mutex_t LOCK_rotation;
   mysql_cond_t COND_rotation;
+  mysql_rwlock_t LOCK_stat_serial;
 
   void vers_init()
   {
     hist_part_id= UINT32_MAX;
     busy_rotation= false;
-    free_parts.empty();
-    free_parts_init= true;
+    stat_trx= NULL;
+    stat_serial= 0;
     mysql_mutex_init(key_TABLE_SHARE_LOCK_rotation, &LOCK_rotation, MY_MUTEX_INIT_FAST);
     mysql_cond_init(key_TABLE_SHARE_COND_rotation, &COND_rotation, NULL);
+    mysql_rwlock_init(key_rwlock_LOCK_stat_serial, &LOCK_stat_serial);
   }
 
   void vers_destroy()
   {
     mysql_mutex_destroy(&LOCK_rotation);
     mysql_cond_destroy(&COND_rotation);
+    mysql_rwlock_destroy(&LOCK_stat_serial);
   }
 
   Field *vers_start_field()
@@ -784,12 +790,6 @@ struct TABLE_SHARE
     return field[row_end_field];
   }
 
-  void vers_part_rotate()
-  {
-    DBUG_ASSERT(!free_parts.is_empty());
-    hist_part_id= (ulong)(void *)(free_parts.pop());
-  }
-
   void vers_wait_rotation()
   {
     while (busy_rotation)