Replace handler::primary_key_is_clustered() with handler::pk_is_clustering_key()

This was done to both simplify the code and also to be easier to handle
storage engines that are clustered on some other index than the primary
key.

As pk_is_clustering_key() and is_clustering_key now are using only
index_flags, these where removed from all storage engines.

10 files changed, 69 insertions, 75 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 813351c658f..cd2f0d4a36a 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -319,7 +319,6 @@ ha_partition::ha_partition(handlerton *hton, TABLE_SHARE *share,
   m_clone_mem_root= clone_mem_root_arg;
   part_share= clone_arg->part_share;
   m_tot_parts= clone_arg->m_tot_parts;
-  m_pkey_is_clustered= clone_arg->primary_key_is_clustered();
   DBUG_VOID_RETURN;
 }
 
@@ -349,7 +348,6 @@ void ha_partition::init_handler_variables()
   m_reorged_parts= 0;
   m_added_file= NULL;
   m_tot_parts= 0;
-  m_pkey_is_clustered= 0;
   m_part_spec.start_part= NO_CURRENT_PART_ID;
   m_scan_value= 2;
   m_ref_length= 0;
@@ -504,8 +502,7 @@ ha_partition::~ha_partition()
      The flag HA_READ_ORDER will be reset for the time being to indicate no
      ordered output is available from partition handler indexes. Later a merge
      sort will be performed using the underlying handlers.
-  5) primary_key_is_clustered and has_transactions are
-     calculated here.
+  5) has_transactions are calculated here.
 
 */
 
@@ -540,19 +537,15 @@ bool ha_partition::initialize_partition(MEM_ROOT *mem_root)
     We create all underlying table handlers here. We do it in this special
     method to be able to report allocation errors.
 
-    Set up primary_key_is_clustered and
-    has_transactions since they are called often in all kinds of places,
+    Set up has_transactions since they are called often in all kinds of places,
     other parameters are calculated on demand.
     Verify that all partitions have the same table_flags.
   */
   check_table_flags= m_file[0]->ha_table_flags();
-  m_pkey_is_clustered= TRUE;
   file_array= m_file;
   do
   {
     file= *file_array;
-    if (!file->primary_key_is_clustered())
-      m_pkey_is_clustered= FALSE;
     if (check_table_flags != file->ha_table_flags())
     {
       my_error(ER_MIX_HANDLER_ERROR, MYF(0));
@@ -5434,7 +5427,7 @@ int ha_partition::index_init(uint inx, bool sorted)
   m_ordered= sorted;
   m_ordered_scan_ongoing= FALSE;
   m_curr_key_info[0]= table->key_info+inx;
-  if (m_pkey_is_clustered && table->s->primary_key != MAX_KEY)
+  if (pk_is_clustering_key(table->s->primary_key))
   {
     /*
       if PK is clustered, then the key cmp must use the pk to
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 657eea4cbd5..85cb736b5bd 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -363,7 +363,6 @@ private:
   uint m_rec_length;                     // Local copy of record length
 
   bool m_ordered;                        // Ordered/Unordered index scan
-  bool m_pkey_is_clustered;              // Is primary key clustered
   bool m_create_handler;                 // Handler used to create table
   bool m_is_sub_partitioned;             // Is subpartitioned
   bool m_ordered_scan_ongoing;
@@ -1335,12 +1334,6 @@ public:
   uint min_record_length(uint options) const override;
 
   /*
-    Primary key is clustered can only be true if all underlying handlers have
-    this feature.
-  */
-  bool primary_key_is_clustered() override { return m_pkey_is_clustered; }
-
-  /*
     -------------------------------------------------------------------------
     MODULE compare records
     -------------------------------------------------------------------------
diff --git a/sql/handler.cc b/sql/handler.cc
index c3f78693849..437492fb2bf 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -2813,7 +2813,7 @@ double handler::keyread_time(uint index, uint ranges, ha_rows rows)
 {
   DBUG_ASSERT(ranges == 0 || ranges == 1);
   size_t len= table->key_info[index].key_length + ref_length;
-  if (index == table->s->primary_key && table->file->primary_key_is_clustered())
+  if (table->file->pk_is_clustering_key(index))
     len= table->s->stored_rec_length;
   double cost= (double)rows*len/(stats.block_size+1)*IDX_BLOCK_COPY_COST;
   if (ranges)
diff --git a/sql/handler.h b/sql/handler.h
index 9d159771097..e1a3409f1c9 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -4029,6 +4029,7 @@ public:
   virtual void set_part_info(partition_info *part_info) {return;}
   virtual void return_record_by_parent() { return; }
 
+  /* Information about index. Both index and part starts from 0 */
   virtual ulong index_flags(uint idx, uint part, bool all_parts) const =0;
 
   uint max_record_length() const
@@ -4168,30 +4169,52 @@ public:
   }
 
  /*
-   Check if the primary key (if there is one) is a clustered and a
-   reference key. This means:
+   Check if the key is a clustering key
 
    - Data is stored together with the primary key (no secondary lookup
      needed to find the row data). The optimizer uses this to find out
      the cost of fetching data.
-   - The primary key is part of each secondary key and is used
+
+     Note that in many cases a clustered key is also a reference key.
+     This means that:
+
+   - The key is part of each secondary key and is used
      to find the row data in the primary index when reading trough
      secondary indexes.
    - When doing a HA_KEYREAD_ONLY we get also all the primary key parts
      into the row. This is critical property used by index_merge.
 
    All the above is usually true for engines that store the row
-   data in the primary key index (e.g. in a b-tree), and use the primary
+   data in the primary key index (e.g. in a b-tree), and use the key
    key value as a position().  InnoDB is an example of such an engine.
 
-   For such a clustered primary key, the following should also hold:
+   For a clustered (primary) key, the following should also hold:
    index_flags() should contain HA_CLUSTERED_INDEX
    table_flags() should contain HA_TABLE_SCAN_ON_INDEX
 
+   For a reference key the following should also hold:
+   table_flags() should contain HA_PRIMARY_KEY_IS_READ_INDEX.
+
    @retval TRUE   yes
    @retval FALSE  No.
  */
- virtual bool primary_key_is_clustered() { return FALSE; }
+
+ /* The following code is for primary keys */
+ bool pk_is_clustering_key(uint index) const
+ {
+   /*
+     We have to check for MAX_INDEX as table->s->primary_key can be
+     MAX_KEY in the case where there is no primary key.
+   */
+   return index != MAX_KEY && is_clustering_key(index);
+ }
+ /* Same as before but for other keys, in which case we can skip the check */
+ bool is_clustering_key(uint index) const
+ {
+   DBUG_ASSERT(index != MAX_KEY);
+   return (index_flags(index, 0, 1) & HA_CLUSTERED_INDEX);
+ }
+
  virtual int cmp_ref(const uchar *ref1, const uchar *ref2)
  {
    return memcmp(ref1, ref2, ref_length);
@@ -4921,8 +4944,6 @@ public:
   virtual void update_partition(uint	part_id)
   {}
 
-  virtual bool is_clustering_key(uint index) { return false; }
-
   /**
     Some engines can perform column type conversion with ALGORITHM=INPLACE.
     These functions check for such possibility.
diff --git a/sql/multi_range_read.cc b/sql/multi_range_read.cc
index 0965ec0eb9f..ffc9008a538 100644
--- a/sql/multi_range_read.cc
+++ b/sql/multi_range_read.cc
@@ -74,7 +74,7 @@ handler::multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
                                                             seq_init_param,
                                                             limit);
   uint len= table->key_info[keyno].key_length + table->file->ref_length;
-  if (keyno == table->s->primary_key && table->file->primary_key_is_clustered())
+  if (table->file->is_clustering_key(keyno))
     len= table->s->stored_rec_length;
   /* Assume block is 75 % full */
   uint avg_block_records= ((uint) (table->file->stats.block_size*3/4))/len + 1;
@@ -143,8 +143,7 @@ handler::multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
     cost->reset();
     cost->avg_io_cost= 1; /* assume random seeks */
     cost->idx_avg_io_cost= 1;
-    if (!((keyno == table->s->primary_key && primary_key_is_clustered()) ||
-	   is_clustering_key(keyno)))
+    if (!is_clustering_key(keyno))
     {
       cost->idx_io_count= total_touched_blocks +
 	                  keyread_time(keyno, 0, total_rows);
@@ -225,7 +224,7 @@ ha_rows handler::multi_range_read_info(uint keyno, uint n_ranges, uint n_rows,
   cost->avg_io_cost= 1; /* assume random seeks */
 
   /* Produce the same cost as non-MRR code does */
-  if (!(keyno == table->s->primary_key && primary_key_is_clustered()))
+  if (!pk_is_clustering_key(keyno))
   {
     cost->idx_io_count=  n_ranges + keyread_time(keyno, 0, n_rows);
     cost->cpu_cost= cost->idx_cpu_cost=
@@ -939,7 +938,7 @@ int DsMrr_impl::dsmrr_init(handler *h_arg, RANGE_SEQ_IF *seq_funcs,
   h_idx= (primary_file->inited == handler::INDEX)? primary_file: secondary_file;
   keyno= h_idx->active_index;
 
-  if (!(keyno == table->s->primary_key && h_idx->primary_key_is_clustered()))
+  if (! h_idx->is_clustering_key(keyno))
   {
     strategy= disk_strategy= &reader_factory.ordered_rndpos_reader;
     if (h_arg->pushed_rowid_filter)
@@ -976,11 +975,10 @@ int DsMrr_impl::dsmrr_init(handler *h_arg, RANGE_SEQ_IF *seq_funcs,
     if (strategy != index_strategy)
     {
       uint saved_pk_length=0;
-      if (h_idx->primary_key_is_clustered())
+      uint pk= h_idx->get_table()->s->primary_key;
+      if (h_idx->pk_is_clustering_key(pk))
       {
-        uint pk= h_idx->get_table()->s->primary_key;
-        if (pk != MAX_KEY)
-          saved_pk_length= h_idx->get_table()->key_info[pk].key_length;
+        saved_pk_length= h_idx->get_table()->key_info[pk].key_length;
       }
       
       KEY *used_index= &h_idx->get_table()->key_info[h_idx->active_index];
@@ -1621,8 +1619,7 @@ bool DsMrr_impl::check_cpk_scan(THD *thd, TABLE_SHARE *share, uint keyno,
                                 uint mrr_flags)
 {
   return MY_TEST((mrr_flags & HA_MRR_SINGLE_POINT) &&
-                 keyno == share->primary_key &&
-                 primary_file->primary_key_is_clustered() &&
+                 primary_file->is_clustering_key(keyno) &&
                  optimizer_flag(thd, OPTIMIZER_SWITCH_MRR_SORT_KEYS));
 }
 
@@ -1660,8 +1657,7 @@ bool DsMrr_impl::choose_mrr_impl(uint keyno, ha_rows rows, uint *flags,
   TABLE_SHARE *share= primary_file->get_table_share();
 
   bool doing_cpk_scan= check_cpk_scan(thd, share, keyno, *flags); 
-  bool using_cpk= MY_TEST(keyno == share->primary_key &&
-                          primary_file->primary_key_is_clustered());
+  bool using_cpk= primary_file->is_clustering_key(keyno);
   *flags &= ~HA_MRR_IMPLEMENTATION_FLAGS;
   if (!optimizer_flag(thd, OPTIMIZER_SWITCH_MRR) ||
       *flags & HA_MRR_INDEX_ONLY ||
@@ -1928,7 +1924,7 @@ void get_sweep_read_cost(TABLE *table, ha_rows nrows, bool interrupted,
   DBUG_ENTER("get_sweep_read_cost");
 
   cost->reset();
-  if (table->file->primary_key_is_clustered())
+  if (table->file->pk_is_clustering_key(table->s->primary_key))
   {
     cost->io_count= table->file->read_time(table->s->primary_key,
                                            (uint) nrows, nrows);
diff --git a/sql/opt_index_cond_pushdown.cc b/sql/opt_index_cond_pushdown.cc
index f8d11da1d5e..360ae028f36 100644
--- a/sql/opt_index_cond_pushdown.cc
+++ b/sql/opt_index_cond_pushdown.cc
@@ -353,13 +353,11 @@ void push_index_cond(JOIN_TAB *tab, uint keyno)
   */
   if ((tab->table->file->index_flags(keyno, 0, 1) &
       HA_DO_INDEX_COND_PUSHDOWN) &&
-     optimizer_flag(tab->join->thd, OPTIMIZER_SWITCH_INDEX_COND_PUSHDOWN) &&
-     tab->join->thd->lex->sql_command != SQLCOM_UPDATE_MULTI &&
-     tab->join->thd->lex->sql_command != SQLCOM_DELETE_MULTI &&
-     tab->type != JT_CONST && tab->type != JT_SYSTEM &&
-     !(keyno == tab->table->s->primary_key &&             // (6)
-       tab->table->file->primary_key_is_clustered()))     // (6)
-
+      optimizer_flag(tab->join->thd, OPTIMIZER_SWITCH_INDEX_COND_PUSHDOWN) &&
+      tab->join->thd->lex->sql_command != SQLCOM_UPDATE_MULTI &&
+      tab->join->thd->lex->sql_command != SQLCOM_DELETE_MULTI &&
+      tab->type != JT_CONST && tab->type != JT_SYSTEM &&
+      !tab->table->file->is_clustering_key(keyno)) // 6
   {
     DBUG_EXECUTE("where",
                  print_where(tab->select_cond, "full cond", QT_ORDINARY););
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 8ce89705127..1f15ab59aa6 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1384,8 +1384,7 @@ bool
 QUICK_INDEX_SORT_SELECT::push_quick_back(QUICK_RANGE_SELECT *quick_sel_range)
 {
   DBUG_ENTER("QUICK_INDEX_SORT_SELECT::push_quick_back");
-  if (head->file->primary_key_is_clustered() &&
-      quick_sel_range->index == head->s->primary_key)
+  if (head->file->is_clustering_key(quick_sel_range->index))
   {
    /*
      A quick_select over a clustered primary key is handled specifically
@@ -2556,7 +2555,7 @@ static int fill_used_fields_bitmap(PARAM *param)
   bitmap_union(&param->needed_fields, table->write_set);
 
   pk= param->table->s->primary_key;
-  if (pk != MAX_KEY && param->table->file->primary_key_is_clustered())
+  if (param->table->file->pk_is_clustering_key(pk))
   {
     /* The table uses clustered PK and it is not internally generated */
     KEY_PART_INFO *key_part= param->table->key_info[pk].key_part;
@@ -4901,16 +4900,16 @@ static void dbug_print_singlepoint_range(SEL_ARG **start, uint num)
 double get_sweep_read_cost(const PARAM *param, ha_rows records)
 {
   double result;
+  uint pk= param->table->s->primary_key;
   DBUG_ENTER("get_sweep_read_cost");
-  if (param->table->file->primary_key_is_clustered() ||
+  if (param->table->file->pk_is_clustering_key(pk) ||
       param->table->file->stats.block_size == 0 /* HEAP */)
   {
     /*
       We are using the primary key to find the rows.
       Calculate the cost for this.
     */
-    result= param->table->file->read_time(param->table->s->primary_key,
-                                          (uint)records, records);
+    result= param->table->file->read_time(pk, (uint)records, records);
   }
   else
   {
@@ -5032,7 +5031,6 @@ TABLE_READ_PLAN *get_best_disjunct_quick(PARAM *param, SEL_IMERGE *imerge,
   double imerge_cost= 0.0;
   ha_rows cpk_scan_records= 0;
   ha_rows non_cpk_scan_records= 0;
-  bool pk_is_clustered= param->table->file->primary_key_is_clustered();
   bool all_scans_ror_able= TRUE;
   bool all_scans_rors= TRUE;
   uint unique_calc_buff_size;
@@ -5102,9 +5100,7 @@ TABLE_READ_PLAN *get_best_disjunct_quick(PARAM *param, SEL_IMERGE *imerge,
     imerge_cost += (*cur_child)->read_cost;
     all_scans_ror_able &= ((*ptree)->n_ror_scans > 0);
     all_scans_rors &= (*cur_child)->is_ror;
-    if (pk_is_clustered &&
-        param->real_keynr[(*cur_child)->key_idx] ==
-        param->table->s->primary_key)
+    if (param->table->file->is_clustering_key(param->real_keynr[(*cur_child)->key_idx]))
     {
       cpk_scan= cur_child;
       cpk_scan_records= (*cur_child)->records;
@@ -5686,14 +5682,14 @@ bool prepare_search_best_index_intersect(PARAM *param,
   common->table_cardinality= 
     get_table_cardinality_for_index_intersect(table);
 
-  if (table->file->primary_key_is_clustered())
+  if (table->file->ha_table_flags() & HA_TABLE_SCAN_ON_INDEX)
   {
     INDEX_SCAN_INFO **index_scan_end;
     index_scan= tree->index_scans;
     index_scan_end= index_scan+n_index_scans;
     for ( ; index_scan < index_scan_end; index_scan++)
     {  
-      if ((*index_scan)->keynr == table->s->primary_key)
+      if (table->file->is_clustering_key((*index_scan)->keynr))
       {
         common->cpk_scan= cpk_scan= *index_scan;
         break;
@@ -6975,7 +6971,8 @@ TRP_ROR_INTERSECT *get_best_ror_intersect(const PARAM *param, SEL_TREE *tree,
                                                      sizeof(ROR_SCAN_INFO*)*
                                                      param->keys)))
     return NULL;
-  cpk_no= ((param->table->file->primary_key_is_clustered()) ?
+  cpk_no= (param->table->file->
+           pk_is_clustering_key(param->table->s->primary_key) ?
            param->table->s->primary_key : MAX_KEY);
 
   for (idx= 0, cur_ror_scan= tree->ror_scans; idx < param->keys; idx++)
@@ -11061,7 +11058,6 @@ ha_rows check_quick_select(PARAM *param, uint idx, bool index_only,
   */
   *mrr_flags|= HA_MRR_NO_ASSOCIATION | HA_MRR_SORTED;
 
-  bool pk_is_clustered= file->primary_key_is_clustered();
   // TODO: param->max_key_parts holds 0 now, and not the #keyparts used.
   // Passing wrong second argument to index_flags() makes no difference for
   // most storage engines but might be an issue for MyRocks with certain
@@ -11105,7 +11101,7 @@ ha_rows check_quick_select(PARAM *param, uint idx, bool index_only,
         MY_MIN(param->table->quick_condition_rows, rows);
       param->table->quick_rows[keynr]= rows;
       param->table->quick_costs[keynr]= cost->total_cost();
-      if (keynr == param->table->s->primary_key && pk_is_clustered)
+      if (param->table->file->is_clustering_key(keynr))
 	param->table->quick_index_only_costs[keynr]= 0;
       else
         param->table->quick_index_only_costs[keynr]= cost->index_only_cost();
@@ -11122,7 +11118,7 @@ ha_rows check_quick_select(PARAM *param, uint idx, bool index_only,
     */
     seq.is_ror_scan= FALSE;
   }
-  else if (param->table->s->primary_key == keynr && pk_is_clustered)
+  else if (param->table->file->is_clustering_key(keynr))
   {
     /* Clustered PK scan is always a ROR scan (TODO: same as above) */
     seq.is_ror_scan= TRUE;
@@ -11207,7 +11203,7 @@ static bool is_key_scan_ror(PARAM *param, uint keynr, uint8 nparts)
 
   key_part= table_key->key_part + nparts;
   pk_number= param->table->s->primary_key;
-  if (!param->table->file->primary_key_is_clustered() || pk_number == MAX_KEY)
+  if (!param->table->file->pk_is_clustering_key(pk_number))
     return FALSE;
 
   KEY_PART_INFO *pk_part= param->table->key_info[pk_number].key_part;
@@ -13650,8 +13646,7 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree, double read_time)
   /*
     Check (SA6) if clustered key is used
   */
-  if (is_agg_distinct && index == table->s->primary_key &&
-      table->file->primary_key_is_clustered())
+  if (is_agg_distinct && table->file->is_clustering_key(index))
   {
     trace_group.add("usable", false)
                .add("cause", "index is clustered");
diff --git a/sql/rowid_filter.cc b/sql/rowid_filter.cc
index 865f22b431a..d6c633aa3e2 100644
--- a/sql/rowid_filter.cc
+++ b/sql/rowid_filter.cc
@@ -356,7 +356,7 @@ void TABLE::init_cost_info_for_usable_range_rowid_filters(THD *thd)
   {
     if (!(file->index_flags(key_no, 0, 1) & HA_DO_RANGE_FILTER_PUSHDOWN))  // !1
       continue;
-    if (key_no == s->primary_key && file->primary_key_is_clustered())      // !2
+    if (file->is_clustering_key(key_no))                              // !2
       continue;
    if (quick_rows[key_no] >
        get_max_range_rowid_filter_elems_for_table(thd, this,
@@ -447,7 +447,7 @@ TABLE::best_range_rowid_filter_for_partial_join(uint access_key_no,
     clustered primary key it would, but the current InnoDB code does not
     allow it. Later this limitation will be lifted
   */
-  if (access_key_no == s->primary_key && file->primary_key_is_clustered())
+  if (file->is_clustering_key(access_key_no))
     return 0;
 
   Range_rowid_filter_cost_info *best_filter= 0;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 40517a6f343..24e158694c5 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -13151,8 +13151,7 @@ make_join_readinfo(JOIN *join, ulonglong options, uint no_jbuf_after)
                 See bug #26447: "Using the clustered index for a table scan
                 is always faster than using a secondary index".
               */
-              if (table->s->primary_key != MAX_KEY &&
-                  table->file->primary_key_is_clustered())
+              if (table->file->pk_is_clustering_key(table->s->primary_key))
                 tab->index= table->s->primary_key;
               else
 #endif
@@ -22860,7 +22859,7 @@ static int test_if_order_by_key(JOIN *join,
   if (have_pk_suffix && reverse == -1)
   {
     uint pk_parts= table->key_info[pk].user_defined_key_parts;
-    if (!(table->file->index_flags(pk, pk_parts, 1) & HA_READ_PREV))
+    if (!(table->file->index_flags(pk, pk_parts-1, 1) & HA_READ_PREV))
       reverse= 0;                               // Index can't be used
   }
 
@@ -23540,8 +23539,7 @@ check_reverse_order:
           if the table is accessed by the primary key
         */
         if (tab->rowid_filter &&
-            tab->index == table->s->primary_key &&
-            table->file->primary_key_is_clustered())
+            table->file->is_clustering_key(tab->index))
 	{
           tab->range_rowid_filter_info= 0;
           delete tab->rowid_filter;
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 5a4bf5ac07f..c2afaaef13c 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1499,8 +1499,8 @@ bool unsafe_key_update(List<TABLE_LIST> leaves, table_map tables_for_update)
     if (!tl->is_jtbm() && (tl->table->map & tables_for_update))
     {
       TABLE *table1= tl->table;
-      bool primkey_clustered= (table1->file->primary_key_is_clustered() &&
-                               table1->s->primary_key != MAX_KEY);
+      bool primkey_clustered= (table1->file->
+                               pk_is_clustering_key(table1->s->primary_key));
 
       bool table_partitioned= false;
 #ifdef WITH_PARTITION_STORAGE_ENGINE