Improve logic to include originalUniqueKey in sharedUniqueKeys (#1453)

Until now, when building the list of SharedUniqueKeys, the code was selecting only the originalUniqueKeys that existed in ghostUniqueKeys list.

However, as part of the alter table, it is possible to add more unique keys or even composed primary keys with multiple columns.

The new logic keeps the old behaviour (it matches originalUniqueKey with exactly the same ghostUniqueKey) but also supports the cases where the originalUniqueKey is now a subset of one of the new ghostUniqueKeys.

If such a case happens, we can still use the originalUniqueKey as normal because a new ghostUniqueKey with the columns of originalUniqueKey plus some new columns is inherently unique just by the columns in the originalUniqueKey, no matter the value in the new columns of the new unique key.

go/logic/inspect.go

@@ -779,8 +779,9 @@ func (this *Inspector) getSharedUniqueKeys(originalUniqueKeys, ghostUniqueKeys [
 	// the ALTER is on the name itself...
 	for _, originalUniqueKey := range originalUniqueKeys {
 		for _, ghostUniqueKey := range ghostUniqueKeys {
-			if originalUniqueKey.Columns.EqualsByNames(&ghostUniqueKey.Columns) {
+			if originalUniqueKey.Columns.IsSubsetOf(&ghostUniqueKey.Columns) {
 				uniqueKeys = append(uniqueKeys, originalUniqueKey)
+				break
 			}
 		}
 	}

go/logic/inspect_test.go

@@ -16,6 +16,7 @@ func TestInspectGetSharedUniqueKeys(t *testing.T) {
 	origUniqKeys := []*sql.UniqueKey{
 		{Columns: *sql.NewColumnList([]string{"id", "item_id"})},
 		{Columns: *sql.NewColumnList([]string{"id", "org_id"})},
+		{Columns: *sql.NewColumnList([]string{"id"})},
 	}
 	ghostUniqKeys := []*sql.UniqueKey{
 		{Columns: *sql.NewColumnList([]string{"id", "item_id"})},
@@ -24,7 +25,8 @@ func TestInspectGetSharedUniqueKeys(t *testing.T) {
 	}
 	inspector := &Inspector{}
 	sharedUniqKeys := inspector.getSharedUniqueKeys(origUniqKeys, ghostUniqKeys)
-	require.Len(t, sharedUniqKeys, 2)
+	require.Len(t, sharedUniqKeys, 3)
 	require.Equal(t, "id,item_id", sharedUniqKeys[0].Columns.String())
 	require.Equal(t, "id,org_id", sharedUniqKeys[1].Columns.String())
+	require.Equal(t, "id", sharedUniqKeys[2].Columns.String())
 }

localtests/fail-no-shared-uk/create.sql

@@ -1,10 +1,10 @@
 drop table if exists gh_ost_test;
 create table gh_ost_test (
-  id int auto_increment,
+  id int not null,
   i int not null,
   ts timestamp,
   primary key(id)
-) auto_increment=1;
+);
 
 drop event if exists gh_ost_test;
 delimiter ;;

localtests/fail-no-shared-uk/extra_args

@@ -1 +1 @@
---alter="drop primary key, add primary key (id, i)" 
+--alter="drop primary key, add primary key (i)" 

localtests/shared-uk/create.sql

@@ -0,0 +1,22 @@
+drop table if exists gh_ost_test;
+create table gh_ost_test (
+  id int auto_increment,
+  i int not null,
+  ts timestamp,
+  primary key(id)
+) auto_increment=1;
+
+drop event if exists gh_ost_test;
+delimiter ;;
+create event gh_ost_test
+  on schedule every 1 second
+  starts current_timestamp
+  ends current_timestamp + interval 60 second
+  on completion not preserve
+  enable
+  do
+begin
+  insert into gh_ost_test values (null, 11, now());
+  insert into gh_ost_test values (null, 13, now());
+  insert into gh_ost_test values (null, 17, now());
+end ;;

localtests/shared-uk/extra_args

@@ -0,0 +1 @@
+--alter="drop primary key, add primary key (id, i)"