MDEV-29069 follow-up: support partially usable keys

4 files changed, 29 insertions, 19 deletions

diff --git a/mysql-test/main/alter_table_online_debug.result b/mysql-test/main/alter_table_online_debug.result
index ca10c410b40..7c617dbe271 100644
--- a/mysql-test/main/alter_table_online_debug.result
+++ b/mysql-test/main/alter_table_online_debug.result
@@ -1011,8 +1011,8 @@ update t set a = a + 1 where a = 10;
 set debug_sync= 'now signal goforit';
 connection default;
 Warnings:
-Note	1105	Key chosen: -1
-Note	1105	Key chosen: -1
+Note	1105	Key chosen: 0
+Note	1105	Key chosen: 0
 select a from t;
 a
 11
diff --git a/sql/log_event.cc b/sql/log_event.cc
index d4d672e1dca..0e795689a7e 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3272,7 +3272,7 @@ Rows_log_event::Rows_log_event(const uchar *buf, uint event_len,
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
     , m_curr_row(NULL), m_curr_row_end(NULL),
     m_key(NULL), m_key_info(NULL), m_key_nr(0),
-    master_had_triggers(0)
+    m_usable_key_parts(0), master_had_triggers(0)
 #endif
 {
   DBUG_ENTER("Rows_log_event::Rows_log_event(const char*,...)");
diff --git a/sql/log_event.h b/sql/log_event.h
index a8ff4a34802..a93c6f2430d 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -5262,9 +5262,10 @@ protected:
   uchar    *m_key;      /* Buffer to keep key value during searches */
   KEY      *m_key_info; /* Pointer to KEY info for m_key_nr */
   uint      m_key_nr;   /* Key number */
+  uint      m_usable_key_parts; /* A number of key_parts suited to lookup */
   bool master_had_triggers;     /* set after tables opening */
 
-  bool key_suits_event(const KEY *key) const;
+  uint key_parts_suit_event(const KEY *key) const;
 
   int find_key(); // Find a best key to use in find_row()
   int find_row(rpl_group_info *);
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 31d0975799c..473feb9f6dd 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -8065,10 +8065,11 @@ record_compare_exit:
   Basically we exclude all the default-filled fields based on
   has_explicit_value bitmap.
  */
-bool Rows_log_event::key_suits_event(const KEY *key) const
+uint Rows_log_event::key_parts_suit_event(const KEY *key) const
 {
   uint master_columns= m_online_alter ? 0 : m_cols.n_bits;
-  for (uint p= 0; p < key->ext_key_parts; p++)
+  uint p= 0;
+  for (;p < key->ext_key_parts; p++)
   {
     Field *f= key->key_part[p].field;
     uint non_deterministic_default= f->default_value &&
@@ -8080,9 +8081,9 @@ bool Rows_log_event::key_suits_event(const KEY *key) const
     if (f->field_index >= master_columns
         && !bitmap_is_set(&m_table->has_value_set, f->field_index)
         && (non_deterministic_default || next_number_field == f->field_index))
-      return false;
+      break;
   }
-  return true;
+  return p;
 }
 
 /**
@@ -8098,7 +8099,7 @@ bool Rows_log_event::key_suits_event(const KEY *key) const
 */
 int Rows_log_event::find_key()
 {
-  uint i, best_key_nr, last_part;
+  uint i, best_key_nr;
   KEY *key, *UNINIT_VAR(best_key);
   ulong UNINIT_VAR(best_rec_per_key), tmp;
   DBUG_ENTER("Rows_log_event::find_key");
@@ -8117,6 +8118,7 @@ int Rows_log_event::find_key()
 
   m_key_nr= MAX_KEY;
   best_key_nr= MAX_KEY;
+  uint parts_suit= 0;
 
   /*
     Keys are sorted so that any primary key is first, followed by unique keys,
@@ -8127,35 +8129,38 @@ int Rows_log_event::find_key()
   {
     if (!m_table->s->keys_in_use.is_set(i))
       continue;
-    if (!key_suits_event(key))
+    parts_suit= key_parts_suit_event(key);
+    if (!parts_suit)
       continue;
     /*
       We cannot use a unique key with NULL-able columns to uniquely identify
       a row (but we can still select it for range scan below if nothing better
       is available).
     */
-    if ((key->flags & (HA_NOSAME | HA_NULL_PART_KEY)) == HA_NOSAME)
+    if ((key->flags & (HA_NOSAME | HA_NULL_PART_KEY)) == HA_NOSAME &&
+        parts_suit == key->user_defined_key_parts)
     {
       best_key_nr= i;
       best_key= key;
+      m_usable_key_parts= parts_suit;
       break;
     }
     /*
       We can only use a non-unique key if it allows range scans (ie. skip
       FULLTEXT indexes and such).
     */
-    last_part= key->user_defined_key_parts - 1;
     DBUG_PRINT("info", ("Index %s rec_per_key[%u]= %lu",
-                        key->name.str, last_part, key->rec_per_key[last_part]));
-    if (!(m_table->file->index_flags(i, last_part, 1) & HA_READ_NEXT))
+                        key->name.str, parts_suit, key->rec_per_key[parts_suit]));
+    if (!(m_table->file->index_flags(i, parts_suit, 1) & HA_READ_NEXT))
       continue;
 
-    tmp= key->rec_per_key[last_part];
+    tmp= key->rec_per_key[parts_suit - 1];
     if (best_key_nr == MAX_KEY || (tmp > 0 && tmp < best_rec_per_key))
     {
       best_key_nr= i;
       best_key= key;
       best_rec_per_key= tmp;
+      m_usable_key_parts= parts_suit;
     }
   }
 
@@ -8310,7 +8315,8 @@ int Rows_log_event::find_row(rpl_group_info *rgi)
 
   if ((table->file->ha_table_flags() & HA_PRIMARY_KEY_REQUIRED_FOR_POSITION) &&
       table->s->primary_key < MAX_KEY &&
-      m_key_nr == table->s->primary_key)
+      m_key_nr == table->s->primary_key &&
+      m_usable_key_parts == table->key_info[table->s->primary_key].ext_key_parts)
   {
     /*
       Use a more efficient method to fetch the record given by
@@ -8398,7 +8404,8 @@ int Rows_log_event::find_row(rpl_group_info *rgi)
       table->record[0][table->s->null_bytes - 1]|=
         256U - (1U << table->s->last_null_bit_pos);
 
-    error= table->file->ha_index_read_map(table->record[0], m_key, HA_WHOLE_KEY,
+    error= table->file->ha_index_read_map(table->record[0], m_key,
+                                          make_keypart_map(m_usable_key_parts),
                                           HA_READ_KEY_EXACT);
     if (unlikely(error))
     {
@@ -8587,14 +8594,16 @@ Delete_rows_log_event::do_before_row_operations(const rpl_group_info *rgi)
   unpack_row(rgi, m_table, m_width, m_curr_row, &m_cols,
              &curr_row_end, &m_master_reclength, m_rows_end);
 
+  KEY *pk_info= m_table->key_info + m_table->s->primary_key;
   if ((m_table->file->ha_table_flags() & HA_PRIMARY_KEY_REQUIRED_FOR_POSITION) &&
-      m_table->s->primary_key < MAX_KEY
-      && key_suits_event(&m_table->key_info[m_table->s->primary_key]))
+      m_table->s->primary_key < MAX_KEY &&
+      key_parts_suit_event(pk_info) == pk_info->ext_key_parts)
   {
     /*
       We don't need to allocate any memory for m_key since it is not used.
     */
     m_key_nr= m_table->s->primary_key;
+    m_usable_key_parts= pk_info->ext_key_parts;
     return 0;
   }
   if (do_invoke_trigger())