BUG#53893: RBR: nullable unique key can lead to out-of-sync slave

When using Unique Keys with nullable parts in RBR, the slave can
choose the wrong row to update. This happens because a table with
an unique key containing nullable parts cannot strictly guarantee 
uniqueness. As stated in the manual, for all engines, a UNIQUE 
index allows multiple NULL values for columns that can contain 
NULL.

We fix this at the slave by extending the checks before assuming
that the row found through an unique index is is the correct
one. This means that when a record (R) is fetched from the storage
engine and a key that is not primary (K) is used, the server does 
the following: 

 - If K is unique and has no nullable parts, it returns R;
 - Otherwise, if any field in the before image that is part of K
   is null do an index scan;
 - If there is no NULL field in the BI part of K, then return R.

A side change: renamed the existing test case file and added a
test case covering the changes in this patch.

1 files changed, 32 insertions, 2 deletions

diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index 60b0df5253a..5f2d55d6477 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -2428,8 +2428,38 @@ int Old_rows_log_event::find_row(const Relay_log_info *rli)
     */
     if (table->key_info->flags & HA_NOSAME)
     {
-      table->file->ha_index_end();
-      DBUG_RETURN(0);
+      /* Unique does not have non nullable part */
+      if (!(table->key_info->flags & (HA_NULL_PART_KEY)))
+      {
+        table->file->ha_index_end();
+        DBUG_RETURN(0);
+      }
+      else
+      {
+        KEY *keyinfo= table->key_info;
+        /*
+          Unique has nullable part. We need to check if there is any field in the
+          BI image that is null and part of UNNI.
+        */
+        bool null_found= FALSE;
+
+        for (uint i=0, fieldnr= keyinfo->key_part[i].fieldnr - 1 ;
+             (i < keyinfo->key_parts) && !null_found ;
+             i++, fieldnr= keyinfo->key_part[i].fieldnr - 1)
+        {
+          Field **f= table->field+fieldnr;
+          if ((*f)->is_null())
+            null_found= TRUE;
+        }
+
+        if (!null_found)
+        {
+          table->file->ha_index_end();
+          DBUG_RETURN(0);
+        }
+
+        /* else fall through to index scan */
+      }
     }
 
     /*