Optimize reactions table (#13406)

* Optimize reactions table

Change reactions primary key to (PostId, UserId, EmojiName) so fetching
reactions for post will use primary key lookup instead of table scan.

* fix db version

* review fixes

* update database schema in scripts/

scripts/mattermost-mysql-5.0.sql

@@ -692,7 +692,7 @@ CREATE TABLE `Reactions` (
   `PostId` varchar(26) NOT NULL,
   `EmojiName` varchar(64) NOT NULL,
   `CreateAt` bigint(20) DEFAULT NULL,
-  PRIMARY KEY (`UserId`,`PostId`,`EmojiName`)
+  PRIMARY KEY (`PostId`,`UserId`,`EmojiName`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
 /*!40101 SET character_set_client = @saved_cs_client */;
 

scripts/mattermost-postgresql-5.0.sql

@@ -1080,7 +1080,7 @@ ALTER TABLE ONLY public.preferences
 --
 
 ALTER TABLE ONLY public.reactions
-    ADD CONSTRAINT reactions_pkey PRIMARY KEY (userid, postid, emojiname);
+    ADD CONSTRAINT reactions_pkey PRIMARY KEY (postid, userid, emojiname);
 
 
 --

store/sqlstore/store.go

@@ -60,6 +60,7 @@ type SqlStore interface {
 	GetMaxLengthOfColumnIfExists(tableName string, columnName string) string
 	AlterColumnTypeIfExists(tableName string, columnName string, mySqlColType string, postgresColType string) bool
 	AlterColumnDefaultIfExists(tableName string, columnName string, mySqlColDefault *string, postgresColDefault *string) bool
+	AlterPrimaryKey(tableName string, columnNames []string) bool
 	CreateUniqueIndexIfNotExists(indexName string, tableName string, columnName string) bool
 	CreateIndexIfNotExists(indexName string, tableName string, columnName string) bool
 	CreateCompositeIndexIfNotExists(indexName string, tableName string, columnNames []string) bool

store/sqlstore/supplier.go

@@ -64,6 +64,7 @@ const (
 	EXIT_REMOVE_INDEX_SQLITE         = 136
 	EXIT_TABLE_EXISTS_SQLITE         = 137
 	EXIT_DOES_COLUMN_EXISTS_SQLITE   = 138
+	EXIT_ALTER_PRIMARY_KEY           = 139
 )
 
 type SqlSupplierStores struct {
@@ -755,6 +756,66 @@ func (ss *SqlSupplier) AlterColumnDefaultIfExists(tableName string, columnName s
 	return true
 }
 
+func (ss *SqlSupplier) AlterPrimaryKey(tableName string, columnNames []string) bool {
+	var currentPrimaryKey string
+	var err error
+	// get the current primary key as a comma separated list of columns
+	if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
+		query := `
+			SELECT GROUP_CONCAT(column_name ORDER BY seq_in_index) AS PK
+		FROM
+			information_schema.statistics
+		WHERE
+			table_schema = DATABASE()
+		AND table_name = ?
+		AND index_name = 'PRIMARY'
+		GROUP BY
+			index_name`
+		currentPrimaryKey, err = ss.GetMaster().SelectStr(query, tableName)
+	} else if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
+		query := `
+			SELECT string_agg(a.attname, ',') AS pk
+		FROM
+			pg_constraint AS c
+		CROSS JOIN LATERAL
+			UNNEST(c.conkey) AS cols(colnum)
+		INNER JOIN
+			pg_attribute AS a ON a.attrelid = c.conrelid
+		AND cols.colnum = a.attnum
+		WHERE
+			c.contype = 'p'
+		AND c.conrelid = '` + strings.ToLower(tableName) + `'::REGCLASS`
+		currentPrimaryKey, err = ss.GetMaster().SelectStr(query)
+	} else if ss.DriverName() == model.DATABASE_DRIVER_SQLITE {
+		// SQLite doesn't support altering primary key
+		return true
+	}
+	if err != nil {
+		mlog.Critical("Failed to get current primary key", mlog.String("table", tableName), mlog.Err(err))
+		time.Sleep(time.Second)
+		os.Exit(EXIT_ALTER_PRIMARY_KEY)
+	}
+
+	primaryKey := strings.Join(columnNames, ",")
+	if strings.EqualFold(currentPrimaryKey, primaryKey) {
+		return false
+	}
+	// alter primary key
+	var alterQuery string
+	if ss.DriverName() == model.DATABASE_DRIVER_MYSQL {
+		alterQuery = "ALTER TABLE " + tableName + " DROP PRIMARY KEY, ADD PRIMARY KEY (" + primaryKey + ")"
+	} else if ss.DriverName() == model.DATABASE_DRIVER_POSTGRES {
+		alterQuery = "ALTER TABLE " + tableName + " DROP CONSTRAINT " + strings.ToLower(tableName) + "_pkey, ADD PRIMARY KEY (" + strings.ToLower(primaryKey) + ")"
+	}
+	_, err = ss.GetMaster().ExecNoTimeout(alterQuery)
+	if err != nil {
+		mlog.Critical("Failed to alter primary key", mlog.String("table", tableName), mlog.Err(err))
+		time.Sleep(time.Second)
+		os.Exit(EXIT_ALTER_PRIMARY_KEY)
+	}
+	return true
+}
+
 func (ss *SqlSupplier) CreateUniqueIndexIfNotExists(indexName string, tableName string, columnName string) bool {
 	return ss.createIndexIfNotExists(indexName, tableName, []string{columnName}, INDEX_TYPE_DEFAULT, true)
 }

store/sqlstore/supplier_reactions.go

@@ -20,7 +20,7 @@ func newSqlReactionStore(sqlStore SqlStore) store.ReactionStore {
 	s := &SqlReactionStore{sqlStore}
 
 	for _, db := range sqlStore.GetAllConns() {
-		table := db.AddTableWithName(model.Reaction{}, "Reactions").SetKeys(false, "UserId", "PostId", "EmojiName")
+		table := db.AddTableWithName(model.Reaction{}, "Reactions").SetKeys(false, "PostId", "UserId", "EmojiName")
 		table.ColMap("UserId").SetMaxSize(26)
 		table.ColMap("PostId").SetMaxSize(26)
 		table.ColMap("EmojiName").SetMaxSize(64)

store/sqlstore/upgrade.go

@@ -753,7 +753,6 @@ func upgradeDatabaseToVersion519(sqlStore SqlStore) {
 
 func upgradeDatabaseToVersion520(sqlStore SqlStore) {
 	if shouldPerformUpgrade(sqlStore, VERSION_5_19_0, VERSION_5_20_0) {
-
 		sqlStore.CreateColumnIfNotExistsNoDefault("Bots", "LastIconUpdate", "bigint", "bigint")
 
 		sqlStore.CreateColumnIfNotExists("GroupTeams", "SchemeAdmin", "boolean", "boolean", "0")
@@ -775,6 +774,8 @@ func upgradeDatabaseToVersion521(sqlStore SqlStore) {
 func upgradeDatabaseToVersion522(sqlStore SqlStore) {
 	// TODO: Uncomment following condition when version 5.22.0 is released
 	// if shouldPerformUpgrade(sqlStore, VERSION_5_21_0, VERSION_5_22_0) {
+	// sqlStore.CreateColumnIfNotExistsNoDefault("Bots", "LastIconUpdate", "bigint", "bigint")
+	// sqlStore.AlterPrimaryKey("Reactions", []string{"PostId", "UserId", "EmojiName"})
 
 	// 	saveSchemaVersion(sqlStore, VERSION_5_22_0)
 	// }

store/storetest/mocks/SqlStore.go

@@ -46,6 +46,20 @@ func (_m *SqlStore) AlterColumnTypeIfExists(tableName string, columnName string,
 	return r0
 }
 
+// AlterPrimaryKey provides a mock function with given fields: tableName, columnNames
+func (_m *SqlStore) AlterPrimaryKey(tableName string, columnNames []string) bool {
+	ret := _m.Called(tableName, columnNames)
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(string, []string) bool); ok {
+		r0 = rf(tableName, columnNames)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
 // Audit provides a mock function with given fields:
 func (_m *SqlStore) Audit() store.AuditStore {
 	ret := _m.Called()