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

MM-17468 - Improving performance of fetching threads (#11980)

Просмотр исходного кода 
fetchThreads parameter support in the API
Этот коммит содержится в:
Eli Yukelzon
2019-09-17 14:37:10 +01:00
коммит произвёл GitHub
родитель 39036ffb30
Коммит b3517eaf2f
14 изменённых файлов: 457 добавлений и 235 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

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

@@ -128,6 +128,12 @@ func (s *SqlPostStore) Save(post *model.Post) (*model.Post, *model.AppError) {
if _, err := s.GetMaster().Exec("UPDATE Posts SET UpdateAt = :UpdateAt WHERE Id = :RootId", map[string]interface{}{"UpdateAt": time, "RootId": post.RootId}); err != nil {
mlog.Error("Error updating Post UpdateAt.", mlog.Err(err))
}
} else {
if count, err := s.GetMaster().SelectInt("SELECT COUNT(*) FROM Posts WHERE RootId = :Id", map[string]interface{}{"Id": post.Id}); err != nil {
mlog.Error(fmt.Sprintf("Error fetching post's thread: %v", err.Error()))
} else {
post.ReplyCount = count
}
}
return post, nil
@@ -266,7 +272,7 @@ func (s *SqlPostStore) GetFlaggedPostsForChannel(userId, channelId string, offse
return pl, nil
}
func (s *SqlPostStore) Get(id string) (*model.PostList, *model.AppError) {
func (s *SqlPostStore) Get(id string, skipFetchThreads bool) (*model.PostList, *model.AppError) {
pl := model.NewPostList()
if len(id) == 0 {
@@ -274,35 +280,40 @@ func (s *SqlPostStore) Get(id string) (*model.PostList, *model.AppError) {
}
var post model.Post
err := s.GetReplica().SelectOne(&post, "SELECT * FROM Posts WHERE Id = :Id AND DeleteAt = 0", map[string]interface{}{"Id": id})
var postFetchQuery string
if skipFetchThreads {
postFetchQuery = "SELECT p.*, (SELECT count(Posts.Id) FROM Posts WHERE Posts.RootId = p.RootId) FROM Posts p WHERE p.Id = :Id AND p.DeleteAt = 0"
} else {
postFetchQuery = "SELECT * FROM Posts WHERE Id = :Id AND DeleteAt = 0"
}
err := s.GetReplica().SelectOne(&post, postFetchQuery, map[string]interface{}{"Id": id})
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPost", "store.sql_post.get.app_error", nil, "id="+id+err.Error(), http.StatusNotFound)
}
pl.AddPost(&post)
pl.AddOrder(id)
if !skipFetchThreads {
rootId := post.RootId
rootId := post.RootId
if rootId == "" {
rootId = post.Id
}
if rootId == "" {
rootId = post.Id
if len(rootId) == 0 {
return nil, model.NewAppError("SqlPostStore.GetPost", "store.sql_post.get.app_error", nil, "root_id="+rootId, http.StatusInternalServerError)
}
var posts []*model.Post
_, err = s.GetReplica().Select(&posts, "SELECT * FROM Posts WHERE (Id = :Id OR RootId = :RootId) AND DeleteAt = 0", map[string]interface{}{"Id": rootId, "RootId": rootId})
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPost", "store.sql_post.get.app_error", nil, "root_id="+rootId+err.Error(), http.StatusInternalServerError)
}
for _, p := range posts {
pl.AddPost(p)
pl.AddOrder(p.Id)
}
}
if len(rootId) == 0 {
return nil, model.NewAppError("SqlPostStore.GetPost", "store.sql_post.get.app_error", nil, "root_id="+rootId, http.StatusInternalServerError)
}
var posts []*model.Post
_, err = s.GetReplica().Select(&posts, "SELECT * FROM Posts WHERE (Id = :Id OR RootId = :RootId) AND DeleteAt = 0", map[string]interface{}{"Id": rootId, "RootId": rootId})
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPost", "store.sql_post.get.app_error", nil, "root_id="+rootId+err.Error(), http.StatusInternalServerError)
}
for _, p := range posts {
pl.AddPost(p)
pl.AddOrder(p.Id)
}
return pl, nil
}
@@ -445,14 +456,14 @@ func (s *SqlPostStore) PermanentDeleteByChannel(channelId string) *model.AppErro
return nil
}
func (s *SqlPostStore) GetPosts(channelId string, offset int, limit int, allowFromCache bool) (*model.PostList, *model.AppError) {
if limit > 1000 {
return nil, model.NewAppError("SqlPostStore.GetLinearPosts", "store.sql_post.get_posts.app_error", nil, "channelId="+channelId, http.StatusBadRequest)
func (s *SqlPostStore) GetPosts(options model.GetPostsOptions, allowFromCache bool) (*model.PostList, *model.AppError) {
if options.PerPage > 1000 {
return nil, model.NewAppError("SqlPostStore.GetLinearPosts", "store.sql_post.get_posts.app_error", nil, "channelId="+options.ChannelId, http.StatusBadRequest)
}
offset := options.PerPage * options.Page
// Caching only occurs on limits of 30 and 60, the common limits requested by MM clients
if allowFromCache && offset == 0 && (limit == 60 || limit == 30) {
if cacheItem, ok := s.lastPostsCache.Get(fmt.Sprintf("%s%v", channelId, limit)); ok {
if allowFromCache && offset == 0 && (options.PerPage == 60 || options.PerPage == 30) {
if cacheItem, ok := s.lastPostsCache.Get(fmt.Sprintf("%s%v", options.ChannelId, options.PerPage)); ok {
if s.metrics != nil {
s.metrics.IncrementMemCacheHitCounter("Last Posts Cache")
}
@@ -466,13 +477,13 @@ func (s *SqlPostStore) GetPosts(channelId string, offset int, limit int, allowFr
rpc := make(chan store.StoreResult, 1)
go func() {
posts, err := s.getRootPosts(channelId, offset, limit)
posts, err := s.getRootPosts(options.ChannelId, offset, options.PerPage, options.SkipFetchThreads)
rpc <- store.StoreResult{Data: posts, Err: err}
close(rpc)
}()
cpc := make(chan store.StoreResult, 1)
go func() {
posts, err := s.getParentsPosts(channelId, offset, limit)
posts, err := s.getParentsPosts(options.ChannelId, offset, options.PerPage, options.SkipFetchThreads)
cpc <- store.StoreResult{Data: posts, Err: err}
close(cpc)
}()
@@ -505,18 +516,18 @@ func (s *SqlPostStore) GetPosts(channelId string, offset int, limit int, allowFr
list.MakeNonNil()
// Caching only occurs on limits of 30 and 60, the common limits requested by MM clients
if offset == 0 && (limit == 60 || limit == 30) {
s.lastPostsCache.AddWithExpiresInSecs(fmt.Sprintf("%s%v", channelId, limit), list, LAST_POSTS_CACHE_SEC)
if offset == 0 && (options.PerPage == 60 || options.PerPage == 30) {
s.lastPostsCache.AddWithExpiresInSecs(fmt.Sprintf("%s%v", options.ChannelId, options.PerPage), list, LAST_POSTS_CACHE_SEC)
}
return list, err
}
func (s *SqlPostStore) GetPostsSince(channelId string, time int64, allowFromCache bool) (*model.PostList, *model.AppError) {
func (s *SqlPostStore) GetPostsSince(options model.GetPostsSinceOptions, allowFromCache bool) (*model.PostList, *model.AppError) {
if allowFromCache {
// If the last post in the channel's time is less than or equal to the time we are getting posts since,
// we can safely return no posts.
if cacheItem, ok := s.lastPostTimeCache.Get(channelId); ok && cacheItem.(int64) <= time {
if cacheItem, ok := s.lastPostTimeCache.Get(options.ChannelId); ok && cacheItem.(int64) <= options.Time {
if s.metrics != nil {
s.metrics.IncrementMemCacheHitCounter("Last Post Time")
}
@@ -556,19 +567,19 @@ func (s *SqlPostStore) GetPostsSince(channelId string, time int64, allowFromCach
AND ChannelId = :ChannelId
LIMIT 1000) temp_tab))
ORDER BY CreateAt DESC`,
map[string]interface{}{"ChannelId": channelId, "Time": time})
map[string]interface{}{"ChannelId": options.ChannelId, "Time": options.Time})
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPostsSince", "store.sql_post.get_posts_since.app_error", nil, "channelId="+channelId+err.Error(), http.StatusInternalServerError)
return nil, model.NewAppError("SqlPostStore.GetPostsSince", "store.sql_post.get_posts_since.app_error", nil, "channelId="+options.ChannelId+err.Error(), http.StatusInternalServerError)
}
list := model.NewPostList()
latestUpdate := time
latestUpdate := options.Time
for _, p := range posts {
list.AddPost(p)
if p.UpdateAt > time {
if p.UpdateAt > options.Time {
list.AddOrder(p.Id)
}
if latestUpdate < p.UpdateAt {
@@ -576,21 +587,25 @@ func (s *SqlPostStore) GetPostsSince(channelId string, time int64, allowFromCach
}
}
s.lastPostTimeCache.AddWithExpiresInSecs(channelId, latestUpdate, LAST_POST_TIME_CACHE_SEC)
s.lastPostTimeCache.AddWithExpiresInSecs(options.ChannelId, latestUpdate, LAST_POST_TIME_CACHE_SEC)
return list, nil
}
func (s *SqlPostStore) GetPostsBefore(channelId string, postId string, limit int, offset int) (*model.PostList, *model.AppError) {
return s.getPostsAround(channelId, postId, limit, offset, true)
func (s *SqlPostStore) GetPostsBefore(options model.GetPostsOptions) (*model.PostList, *model.AppError) {
return s.getPostsAround(true, options)
}
func (s *SqlPostStore) GetPostsAfter(channelId string, postId string, limit int, offset int) (*model.PostList, *model.AppError) {
return s.getPostsAround(channelId, postId, limit, offset, false)
func (s *SqlPostStore) GetPostsAfter(options model.GetPostsOptions) (*model.PostList, *model.AppError) {
return s.getPostsAround(false, options)
}
func (s *SqlPostStore) getPostsAround(channelId string, postId string, limit int, offset int, before bool) (*model.PostList, *model.AppError) {
var direction, sort string
func (s *SqlPostStore) getPostsAround(before bool, options model.GetPostsOptions) (*model.PostList, *model.AppError) {
offset := options.Page * options.PerPage
var posts, parents []*model.Post
var direction string
var sort string
if before {
direction = "<"
sort = "DESC"
@@ -598,23 +613,29 @@ func (s *SqlPostStore) getPostsAround(channelId string, postId string, limit int
direction = ">"
sort = "ASC"
}
replyCountSubQuery := s.getQueryBuilder().Select("COUNT(Posts.Id)").From("Posts").Where(sq.Expr("p.RootId = '' AND RootId = p.Id"))
query := s.getQueryBuilder().Select("p.*")
if options.SkipFetchThreads {
query = query.Column(sq.Alias(replyCountSubQuery, "ReplyCount"))
}
query = query.From("Posts p").
Where(sq.And{
sq.Expr(`CreateAt `+direction+` (SELECT CreateAt FROM Posts WHERE Id = ?)`, options.PostId),
sq.Eq{"ChannelId": options.ChannelId},
sq.Eq{"DeleteAt": int(0)},
}).
OrderBy("CreateAt " + sort).
Limit(uint64(options.PerPage)).
Offset(uint64(offset))
queryString, args, err := query.ToSql()
var posts, parents []*model.Post
_, err := s.GetReplica().Select(&posts,
`SELECT
*
FROM
Posts
WHERE
CreateAt `+direction+` (SELECT CreateAt FROM Posts WHERE Id = :PostId)
AND ChannelId = :ChannelId
AND DeleteAt = 0
ORDER BY CreateAt `+sort+`
LIMIT :Limit
OFFSET :Offset`,
map[string]interface{}{"ChannelId": channelId, "PostId": postId, "Limit": limit, "Offset": offset})
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPostContext", "store.sql_post.get_posts_around.get.app_error", nil, "channelId="+channelId+err.Error(), http.StatusInternalServerError)
return nil, model.NewAppError("SqlPostStore.GetPostContext", "store.sql_post.get_posts_around.get.app_error", nil, "channelId="+options.ChannelId+err.Error(), http.StatusInternalServerError)
}
_, err = s.GetMaster().Select(&posts, queryString, args...)
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPostContext", "store.sql_post.get_posts_around.get.app_error", nil, "channelId="+options.ChannelId+err.Error(), http.StatusInternalServerError)
}
if len(posts) > 0 {
@@ -625,28 +646,29 @@ func (s *SqlPostStore) getPostsAround(channelId string, postId string, limit int
rootIds = append(rootIds, post.RootId)
}
}
rootQuery := s.getQueryBuilder().Select("p.*")
if options.SkipFetchThreads {
rootQuery = rootQuery.Column(sq.Alias(replyCountSubQuery, "ReplyCount"))
}
rootQuery = rootQuery.From("Posts p").
Where(sq.And{
sq.Or{
sq.Eq{"RootId": rootIds},
sq.Eq{"Id": rootIds},
},
sq.Eq{"ChannelId": options.ChannelId},
sq.Eq{"DeleteAt": 0},
}).
OrderBy("CreateAt DESC")
keys, params := MapStringsToQueryParams(rootIds, "PostId")
params["ChannelId"] = channelId
params["PostId"] = postId
params["Limit"] = limit
params["Offset"] = offset
_, err = s.GetReplica().Select(&parents,
`SELECT
*
FROM
Posts
WHERE
(Id IN `+keys+` OR RootId IN `+keys+`)
AND ChannelId = :ChannelId
AND DeleteAt = 0
ORDER BY CreateAt DESC`,
params)
rootQueryString, rootArgs, err := rootQuery.ToSql()
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPostContext", "store.sql_post.get_posts_around.get_parent.app_error", nil, "channelId="+channelId+err.Error(), http.StatusInternalServerError)
return nil, model.NewAppError("SqlPostStore.GetPostContext", "store.sql_post.get_posts_around.get_parent.app_error", nil, "channelId="+options.ChannelId+err.Error(), http.StatusInternalServerError)
}
_, err = s.GetMaster().Select(&parents, rootQueryString, rootArgs...)
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetPostContext", "store.sql_post.get_posts_around.get_parent.app_error", nil, "channelId="+options.ChannelId+err.Error(), http.StatusInternalServerError)
}
}
@@ -745,20 +767,29 @@ func (s *SqlPostStore) GetPostAfterTime(channelId string, time int64) (*model.Po
return post, nil
}
func (s *SqlPostStore) getRootPosts(channelId string, offset int, limit int) ([]*model.Post, *model.AppError) {
func (s *SqlPostStore) getRootPosts(channelId string, offset int, limit int, skipFetchThreads bool) ([]*model.Post, *model.AppError) {
var posts []*model.Post
_, err := s.GetReplica().Select(&posts, "SELECT * FROM Posts WHERE ChannelId = :ChannelId AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT :Limit OFFSET :Offset", map[string]interface{}{"ChannelId": channelId, "Offset": offset, "Limit": limit})
var fetchQuery string
if skipFetchThreads {
fetchQuery = "SELECT p.*, (SELECT COUNT(Posts.Id) FROM Posts WHERE p.RootId = '' AND Posts.RootId = p.Id) as ReplyCount FROM Posts p WHERE ChannelId = :ChannelId AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT :Limit OFFSET :Offset"
} else {
fetchQuery = "SELECT * FROM Posts WHERE ChannelId = :ChannelId AND DeleteAt = 0 ORDER BY CreateAt DESC LIMIT :Limit OFFSET :Offset"
}
_, err := s.GetReplica().Select(&posts, fetchQuery, map[string]interface{}{"ChannelId": channelId, "Offset": offset, "Limit": limit})
if err != nil {
return nil, model.NewAppError("SqlPostStore.GetLinearPosts", "store.sql_post.get_root_posts.app_error", nil, "channelId="+channelId+err.Error(), http.StatusInternalServerError)
}
return posts, nil
}
func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int) ([]*model.Post, *model.AppError) {
func (s *SqlPostStore) getParentsPosts(channelId string, offset int, limit int, skipFetchThreads bool) ([]*model.Post, *model.AppError) {
var posts []*model.Post
replyCountQuery := ""
if skipFetchThreads {
replyCountQuery = ` ,(SELECT COUNT(Posts.Id) FROM Posts WHERE q2.RootId = '' AND Posts.RootId = q2.Id) as ReplyCount`
}
_, err := s.GetReplica().Select(&posts,
`SELECT
q2.*
`SELECT q2.*`+replyCountQuery+`
FROM
Posts q2
INNER JOIN