mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

MM-43045: minimize JOIN Posts (#19934)

Просмотр исходного кода
Этот коммит содержится в:
Jesse Hallam
2022-04-19 17:06:31 -03:00
коммит произвёл GitHub
родитель 348602cf00
Коммит 5bd223c836
12 изменённых файлов: 465 добавлений и 282 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

181
store/sqlstore/post_store.go
Просмотреть файл

@@ -215,7 +215,7 @@ func (s *SqlPostStore) SaveMultiple(posts []*model.Post) ([]*model.Post, int, er
}
if err = s.updateThreadsFromPosts(transaction, posts); err != nil {
mlog.Warn("Error updating posts, thread update failed", mlog.Err(err))
return nil, -1, errors.Wrap(err, "update thread from posts failed")
}
if err = transaction.Commit(); err != nil {
@@ -290,7 +290,7 @@ func (s *SqlPostStore) populateReplyCount(posts []*model.Post) error {
Select("RootId, COUNT(Id) AS Count").
From("Posts").
Where(sq.Eq{"RootId": rootIds}).
Where(sq.Eq{"DeleteAt": 0}).
Where(sq.Eq{"Posts.DeleteAt": 0}).
GroupBy("RootId")
queryString, args, err := query.ToSql()
@@ -487,7 +487,7 @@ func (s *SqlPostStore) getFlaggedPosts(userId, channelId, teamId string, offset
AND Category = ?
)
CHANNEL_FILTER
AND DeleteAt = 0
AND Posts.DeleteAt = 0
) as A
INNER JOIN Channels as B
ON B.Id = A.ChannelId
@@ -569,8 +569,8 @@ func (s *SqlPostStore) getPostWithCollapsedThreads(id, userID string, opts model
From("Posts").
LeftJoin("Threads ON Threads.PostId = Id").
LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Id AND ThreadMemberships.UserId = ?", userID).
Where(sq.Eq{"DeleteAt": 0}).
Where(sq.Eq{"Id": id}).ToSql()
Where(sq.Eq{"Posts.DeleteAt": 0}).
Where(sq.Eq{"Posts.Id": id}).ToSql()
err := s.GetReplicaX().Get(&post, postFetchQuery, args...)
if err != nil {
@@ -586,8 +586,8 @@ func (s *SqlPostStore) getPostWithCollapsedThreads(id, userID string, opts model
Select("*").
From("Posts").
Where(sq.Eq{
"RootId": id,
"DeleteAt": 0,
"Posts.RootId": id,
"Posts.DeleteAt": 0,
})
var sort string
@@ -873,7 +873,11 @@ func (s *SqlPostStore) Delete(postID string, time int64, deleteByID string) erro
return errors.Wrap(err, "failed to update Posts")
}
err = s.cleanupThreadComments(transaction, postID, id.RootId, id.UserId)
if id.RootId == "" {
err = s.deleteThread(transaction, postID, time)
} else {
err = s.updateThreadAfterReplyDeletion(transaction, id.RootId, id.UserId)
}
if err != nil {
return errors.Wrapf(err, "failed to cleanup Thread with postid=%s", id.RootId)
@@ -933,13 +937,12 @@ func (s *SqlPostStore) permanentDeleteAllCommentByUser(userId string) error {
}
_, err = transaction.Exec("DELETE FROM Posts WHERE UserId = ? AND RootId != ''", userId)
if err != nil {
return errors.Wrapf(err, "failed to delete Posts with userId=%s", userId)
}
for _, ids := range results {
if err = s.cleanupThreadComments(transaction, ids.Id, ids.RootId, userId); err != nil {
if err = s.updateThreadAfterReplyDeletion(transaction, ids.RootId, userId); err != nil {
return err
}
}
@@ -1103,14 +1106,14 @@ func (s *SqlPostStore) getPostsCollapsedThreads(options model.GetPostsOptions) (
postFetchQuery, args, _ := s.getQueryBuilder().
Select(columns...).
From("Posts").
LeftJoin("Threads ON Threads.PostId = Id").
LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Id AND ThreadMemberships.UserId = ?", options.UserId).
Where(sq.Eq{"DeleteAt": 0}).
LeftJoin("Threads ON Threads.PostId = Posts.Id").
LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Posts.Id AND ThreadMemberships.UserId = ?", options.UserId).
Where(sq.Eq{"Posts.DeleteAt": 0}).
Where(sq.Eq{"Posts.ChannelId": options.ChannelId}).
Where(sq.Eq{"RootId": ""}).
Where(sq.Eq{"Posts.RootId": ""}).
Limit(uint64(options.PerPage)).
Offset(uint64(offset)).
OrderBy("CreateAt DESC").ToSql()
OrderBy("Posts.CreateAt DESC").ToSql()
err := s.GetReplicaX().Select(&posts, postFetchQuery, args...)
if err != nil {
@@ -1187,13 +1190,13 @@ func (s *SqlPostStore) getPostsSinceCollapsedThreads(options model.GetPostsSince
postFetchQuery, args, _ := s.getQueryBuilder().
Select(columns...).
From("Posts").
LeftJoin("Threads ON Threads.PostId = Id").
LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Id AND ThreadMemberships.UserId = ?", options.UserId).
Where(sq.Eq{"DeleteAt": 0}).
LeftJoin("Threads ON Threads.PostId = Posts.Id").
LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Posts.Id AND ThreadMemberships.UserId = ?", options.UserId).
Where(sq.Eq{"Posts.DeleteAt": 0}).
Where(sq.Eq{"Posts.ChannelId": options.ChannelId}).
Where(sq.Gt{"UpdateAt": options.Time}).
Where(sq.Eq{"RootId": ""}).
OrderBy("CreateAt DESC").ToSql()
Where(sq.Gt{"Posts.UpdateAt": options.Time}).
Where(sq.Eq{"Posts.RootId": ""}).
OrderBy("Posts.CreateAt DESC").ToSql()
err := s.GetReplicaX().Select(&posts, postFetchQuery, args...)
if err != nil {
@@ -1316,16 +1319,16 @@ func (s *SqlPostStore) GetPostsSinceForSync(options model.GetPostsSinceForSyncOp
query := s.getQueryBuilder().
Select("*").
From("Posts").
Where(sq.Or{sq.Gt{"UpdateAt": cursor.LastPostUpdateAt}, sq.And{sq.Eq{"UpdateAt": cursor.LastPostUpdateAt}, sq.Gt{"Id": cursor.LastPostId}}}).
OrderBy("UpdateAt", "Id").
Where(sq.Or{sq.Gt{"Posts.UpdateAt": cursor.LastPostUpdateAt}, sq.And{sq.Eq{"Posts.UpdateAt": cursor.LastPostUpdateAt}, sq.Gt{"Posts.Id": cursor.LastPostId}}}).
OrderBy("Posts.UpdateAt", "Id").
Limit(uint64(limit))
if options.ChannelId != "" {
query = query.Where(sq.Eq{"ChannelId": options.ChannelId})
query = query.Where(sq.Eq{"Posts.ChannelId": options.ChannelId})
}
if !options.IncludeDeleted {
query = query.Where(sq.Eq{"DeleteAt": 0})
query = query.Where(sq.Eq{"Posts.DeleteAt": 0})
}
if options.ExcludeRemoteId != "" {
@@ -1402,7 +1405,7 @@ func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions
conditions := sq.And{
sq.Expr(`CreateAt `+direction+` (SELECT CreateAt FROM Posts WHERE Id = ?)`, options.PostId),
sq.Eq{"p.ChannelId": options.ChannelId},
sq.Eq{"DeleteAt": int(0)},
sq.Eq{"p.DeleteAt": int(0)},
}
if options.CollapsedThreads {
conditions = append(conditions, sq.Eq{"RootId": ""})
@@ -1415,7 +1418,7 @@ func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions
// Adding ChannelId and DeleteAt order columns
// to let mysql choose the "idx_posts_channel_id_delete_at_create_at" index always.
// See MM-24170.
OrderBy("p.ChannelId", "DeleteAt", "CreateAt "+sort).
OrderBy("p.ChannelId", "p.DeleteAt", "p.CreateAt "+sort).
Limit(uint64(options.PerPage)).
Offset(uint64(offset))
@@ -1448,8 +1451,8 @@ func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions
rootQuery = rootQuery.From("Posts p").
Where(sq.And{
idQuery,
sq.Eq{"ChannelId": options.ChannelId},
sq.Eq{"DeleteAt": 0},
sq.Eq{"p.ChannelId": options.ChannelId},
sq.Eq{"p.DeleteAt": 0},
}).
OrderBy("CreateAt DESC")
@@ -1505,11 +1508,11 @@ func (s *SqlPostStore) getPostIdAroundTime(channelId string, time int64, before
conditions := sq.And{
direction,
sq.Eq{"ChannelId": channelId},
sq.Eq{"DeleteAt": int(0)},
sq.Eq{"Posts.ChannelId": channelId},
sq.Eq{"Posts.DeleteAt": int(0)},
}
if collapsedThreads {
conditions = sq.And{conditions, sq.Eq{"RootId": ""}}
conditions = sq.And{conditions, sq.Eq{"Posts.RootId": ""}}
}
query := s.getQueryBuilder().
Select("Id").
@@ -1518,7 +1521,7 @@ func (s *SqlPostStore) getPostIdAroundTime(channelId string, time int64, before
// Adding ChannelId and DeleteAt order columns
// to let mysql choose the "idx_posts_channel_id_delete_at_create_at" index always.
// See MM-23369.
OrderBy("ChannelId", "DeleteAt", "CreateAt "+sort).
OrderBy("Posts.ChannelId", "Posts.DeleteAt", "Posts.CreateAt "+sort).
Limit(1)
queryString, args, err := query.ToSql()
@@ -1545,9 +1548,9 @@ func (s *SqlPostStore) GetPostAfterTime(channelId string, time int64, collapsedT
table += " USE INDEX(idx_posts_channel_id_delete_at_create_at)"
}
conditions := sq.And{
sq.Gt{"CreateAt": time},
sq.Eq{"ChannelId": channelId},
sq.Eq{"DeleteAt": int(0)},
sq.Gt{"Posts.CreateAt": time},
sq.Eq{"Posts.ChannelId": channelId},
sq.Eq{"Posts.DeleteAt": int(0)},
}
if collapsedThreads {
conditions = sq.And{conditions, sq.Eq{"RootId": ""}}
@@ -1559,7 +1562,7 @@ func (s *SqlPostStore) GetPostAfterTime(channelId string, time int64, collapsedT
// Adding ChannelId and DeleteAt order columns
// to let mysql choose the "idx_posts_channel_id_delete_at_create_at" index always.
// See MM-23369.
OrderBy("ChannelId", "DeleteAt", "CreateAt ASC").
OrderBy("Posts.ChannelId", "Posts.DeleteAt", "Posts.CreateAt ASC").
Limit(1)
queryString, args, err := query.ToSql()
@@ -1581,9 +1584,9 @@ func (s *SqlPostStore) getRootPosts(channelId string, offset int, limit int, ski
posts := []*model.Post{}
var fetchQuery string
if skipFetchThreads {
fetchQuery = "SELECT p.*, (SELECT COUNT(*) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE ChannelId = ? AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT ? OFFSET ?"
fetchQuery = "SELECT p.*, (SELECT COUNT(*) FROM Posts WHERE Posts.RootId = (CASE WHEN p.RootId = '' THEN p.Id ELSE p.RootId END) AND Posts.DeleteAt = 0) as ReplyCount FROM Posts p WHERE p.ChannelId = ? AND p.DeleteAt = 0 ORDER BY p.CreateAt DESC LIMIT ? OFFSET ?"
} else {
fetchQuery = "SELECT * FROM Posts WHERE ChannelId = ? AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT ? OFFSET ?"
fetchQuery = "SELECT * FROM Posts WHERE Posts.ChannelId = ? AND Posts.DeleteAt = 0 ORDER BY Posts.CreateAt DESC LIMIT ? OFFSET ?"
}
err := s.GetReplicaX().Select(&posts, fetchQuery, channelId, limit, offset)
if err != nil {
@@ -1604,13 +1607,13 @@ func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int,
q.RootId
FROM
(SELECT
RootId
Posts.RootId
FROM
Posts
WHERE
ChannelId = ?
AND DeleteAt = 0
ORDER BY CreateAt DESC
Posts.ChannelId = ?
AND Posts.DeleteAt = 0
ORDER BY Posts.CreateAt DESC
LIMIT ? OFFSET ?) q
WHERE q.RootId != ''`
@@ -1639,10 +1642,10 @@ func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int,
From("Posts p").
Where(sq.And{
where,
sq.Eq{"ChannelId": channelId},
sq.Eq{"DeleteAt": 0},
sq.Eq{"p.ChannelId": channelId},
sq.Eq{"p.DeleteAt": 0},
}).
OrderBy("CreateAt")
OrderBy("p.CreateAt")
sql, args, err := query.ToSql()
if err != nil {
@@ -1675,20 +1678,20 @@ func (s *SqlPostStore) getParentsPostsPostgreSQL(channelId string, offset int, l
q3.RootId
FROM
(SELECT
RootId
Posts.RootId
FROM
Posts
WHERE
ChannelId = ?
AND DeleteAt = 0
ORDER BY CreateAt DESC
Posts.ChannelId = ?
AND Posts.DeleteAt = 0
ORDER BY Posts.CreateAt DESC
LIMIT ? OFFSET ?) q3
WHERE q3.RootId != '') q1
ON `+onStatement+`
WHERE
ChannelId = ?
AND DeleteAt = 0
ORDER BY CreateAt`, channelId, limit, offset, channelId)
q2.ChannelId = ?
AND q2.DeleteAt = 0
ORDER BY q2.CreateAt`, channelId, limit, offset, channelId)
if err != nil {
return nil, errors.Wrapf(err, "failed to find Posts with channelId=%s", channelId)
}
@@ -1835,9 +1838,9 @@ func (s *SqlPostStore) search(teamId string, userId string, params *model.Search
"*",
"(SELECT COUNT(*) FROM Posts WHERE Posts.RootId = (CASE WHEN q2.RootId = '' THEN q2.Id ELSE q2.RootId END) AND Posts.DeleteAt = 0) as ReplyCount",
).From("Posts q2").
Where("DeleteAt = 0").
Where(fmt.Sprintf("Type NOT LIKE '%s%%'", model.PostSystemMessagePrefix)).
OrderByClause("CreateAt DESC").
Where("q2.DeleteAt = 0").
Where(fmt.Sprintf("q2.Type NOT LIKE '%s%%'", model.PostSystemMessagePrefix)).
OrderByClause("q2.CreateAt DESC").
Limit(100)
var err error
@@ -1928,11 +1931,11 @@ func (s *SqlPostStore) search(teamId string, userId string, params *model.Search
Where("Id = ChannelId")
if !params.IncludeDeletedChannels {
inQuery = inQuery.Where("DeleteAt = 0")
inQuery = inQuery.Where("Channels.DeleteAt = 0")
}
if !params.SearchWithoutUserId {
inQuery = inQuery.Where("UserId = ?", userId)
inQuery = inQuery.Where("ChannelMembers.UserId = ?", userId)
}
inQuery = s.buildSearchTeamFilterClause(teamId, inQuery)
@@ -2378,10 +2381,10 @@ func (s *SqlPostStore) GetParentsForExportAfter(limit int, afterId string) ([]*m
FROM
Posts
WHERE
Id > ?
AND RootId = ''
AND DeleteAt = 0
ORDER BY Id
Posts.Id > ?
AND Posts.RootId = ''
AND Posts.DeleteAt = 0
ORDER BY Posts.Id
LIMIT ?`,
afterId, limit)
if err != nil {
@@ -2395,7 +2398,7 @@ func (s *SqlPostStore) GetParentsForExportAfter(limit int, afterId string) ([]*m
builder := s.getQueryBuilder().
Select("p1.*, Users.Username as Username, Teams.Name as TeamName, Channels.Name as ChannelName").
FromSelect(sq.Select("*").From("Posts").Where(sq.Eq{"Id": rootIds}), "p1").
FromSelect(sq.Select("*").From("Posts").Where(sq.Eq{"Posts.Id": rootIds}), "p1").
InnerJoin("Channels ON p1.ChannelId = Channels.Id").
InnerJoin("Teams ON Channels.TeamId = Teams.Id").
InnerJoin("Users ON p1.UserId = Users.Id").
@@ -2598,19 +2601,36 @@ func (s *SqlPostStore) permanentDeleteThreads(transaction *sqlxTxWrapper, postId
return nil
}
// Thread cleanup upon post deletion
// if the post is a comment
// reply count is reduced by 1 and,
// the user is removed from participants if the comment deleted is the last reply from said user.
func (s *SqlPostStore) cleanupThreadComments(transaction *sqlxTxWrapper, postId, rootId string, userId string) error {
// deleteThread marks a thread as deleted at the given time.
func (s *SqlPostStore) deleteThread(transaction *sqlxTxWrapper, postId string, deleteAtTime int64) error {
queryString, args, err := s.getQueryBuilder().
Update("Threads").
Set("DeleteAt", deleteAtTime).
Where(sq.Eq{"PostId": postId}).
ToSql()
if err != nil {
return errors.Wrapf(err, "failed to create SQL query to mark thread for root post %s as deleted", postId)
}
_, err = transaction.Exec(queryString, args...)
if err != nil {
return errors.Wrapf(err, "failed to mark thread for root post %s as deleted", postId)
}
return nil
}
// updateThreadAfterReplyDeletion decrements the thread reply count and adjusts the participants
// list as necessary.
func (s *SqlPostStore) updateThreadAfterReplyDeletion(transaction *sqlxTxWrapper, rootId string, userId string) error {
if rootId != "" {
queryString, args, err := s.getQueryBuilder().
Select("COUNT(Id)").
Select("COUNT(Posts.Id)").
From("Posts").
Where(sq.And{
sq.Eq{"RootId": rootId},
sq.Eq{"UserId": userId},
sq.Eq{"DeleteAt": 0},
sq.Eq{"Posts.RootId": rootId},
sq.Eq{"Posts.UserId": userId},
sq.Eq{"Posts.DeleteAt": 0},
}).
ToSql()
@@ -2675,9 +2695,16 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts
return nil
}
threadsByRootsSql, threadsByRootsArgs, _ := s.getQueryBuilder().
Select("*").
Select(
"Threads.PostId",
"Threads.ChannelId",
"Threads.ReplyCount",
"Threads.LastReplyAt",
"Threads.Participants",
"COALESCE(Threads.DeleteAt, 0) AS DeleteAt",
).
From("Threads").
Where(sq.Eq{"PostId": rootIds}).
Where(sq.Eq{"Threads.PostId": rootIds}).
ToSql()
threadsByRoots := []*model.Thread{}
if err := transaction.Select(&threadsByRoots, threadsByRootsSql, threadsByRootsArgs...); err != nil {
@@ -2697,7 +2724,7 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts
}{}
// calculate participants
if err := transaction.Select(&data, "SELECT UserId, MAX(CreateAt) as RepliedAt FROM Posts WHERE RootId=? AND DeleteAt=0 GROUP BY UserId ORDER BY RepliedAt ASC", rootId); err != nil {
if err := transaction.Select(&data, "SELECT Posts.UserId, MAX(Posts.CreateAt) as RepliedAt FROM Posts WHERE Posts.RootId=? AND Posts.DeleteAt=0 GROUP BY Posts.UserId ORDER BY RepliedAt ASC", rootId); err != nil {
return err
}
@@ -2708,13 +2735,13 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts
// calculate reply count
var count int64
err := transaction.Get(&count, "SELECT COUNT(Id) FROM Posts WHERE RootId=? And DeleteAt=0", rootId)
err := transaction.Get(&count, "SELECT COUNT(Posts.Id) FROM Posts WHERE Posts.RootId=? And Posts.DeleteAt=0", rootId)
if err != nil {
return err
}
// calculate last reply at
var lastReplyAt int64
err = transaction.Get(&lastReplyAt, "SELECT COALESCE(MAX(Posts.CreateAt), 0) FROM Posts WHERE RootID=? and DeleteAt=0", rootId)
err = transaction.Get(&lastReplyAt, "SELECT COALESCE(MAX(Posts.CreateAt), 0) FROM Posts WHERE Posts.RootID=? and Posts.DeleteAt=0", rootId)
if err != nil {
return err
}