PLT-7934: Make query for bulk elasticsearch indexing more efficient. (#7664)

3 files changed, 66 insertions, 16 deletions

diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go
index a1b25b5c5..d8f93d2bc 100644
--- a/store/sqlstore/post_store.go
+++ b/store/sqlstore/post_store.go
@@ -1067,29 +1067,37 @@ func (s SqlPostStore) GetPostsByIds(postIds []string) store.StoreChannel {
 	})
 }
 
-func (s SqlPostStore) GetPostsBatchForIndexing(startTime int64, limit int) store.StoreChannel {
+func (s SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) store.StoreChannel {
 	return store.Do(func(result *store.StoreResult) {
 		var posts []*model.PostForIndexing
 		_, err1 := s.GetSearchReplica().Select(&posts,
-			`(SELECT
-			    Posts.*,
-			    Channels.TeamId,
-			    ParentPosts.CreateAt ParentCreateAt
-			FROM
-				Posts
+			`SELECT
+				PostsQuery.*, Channels.TeamId, ParentPosts.CreateAt ParentCreateAt
+			FROM (
+				SELECT
+					*
+				FROM
+					Posts
+				WHERE
+					Posts.CreateAt >= :StartTime
+				AND
+					Posts.CreateAt < :EndTime
+				ORDER BY
+					CreateAt ASC
+				LIMIT
+					1000
+				)
+			AS
+				PostsQuery
 			LEFT JOIN
 				Channels
 			ON
-				Posts.ChannelId = Channels.Id
+				PostsQuery.ChannelId = Channels.Id
 			LEFT JOIN
 				Posts ParentPosts
 			ON
-				Posts.RootId = ParentPosts.Id
-			WHERE
-				Posts.CreateAt >= :StartTime
-			ORDER BY CreateAt ASC
-			LIMIT :NumPosts)`,
-			map[string]interface{}{"StartTime": startTime, "NumPosts": limit})
+				PostsQuery.RootId = ParentPosts.Id`,
+			map[string]interface{}{"StartTime": startTime, "EndTime": endTime, "NumPosts": limit})
 
 		if err1 != nil {
 			result.Err = model.NewAppError("SqlPostStore.GetPostContext", "store.sql_post.get_posts_batch_for_indexing.get.app_error", nil, err1.Error(), http.StatusInternalServerError)
@@ -1122,3 +1130,15 @@ func (s SqlPostStore) PermanentDeleteBatch(endTime int64, limit int64) store.Sto
 		}
 	})
 }
+
+func (s SqlPostStore) GetOldest() store.StoreChannel {
+	return store.Do(func(result *store.StoreResult) {
+		var post model.Post
+		err := s.GetReplica().SelectOne(&post, "SELECT * FROM Posts ORDER BY CreateAt LIMIT 1")
+		if err != nil {
+			result.Err = model.NewAppError("SqlPostStore.GetOldest", "store.sql_post.get.app_error", nil, err.Error(), http.StatusNotFound)
+		}
+
+		result.Data = &post
+	})
+}
diff --git a/store/store.go b/store/store.go
index 5674a05d5..eada8f395 100644
--- a/store/store.go
+++ b/store/store.go
@@ -182,8 +182,9 @@ type PostStore interface {
 	GetPostsCreatedAt(channelId string, time int64) StoreChannel
 	Overwrite(post *model.Post) StoreChannel
 	GetPostsByIds(postIds []string) StoreChannel
-	GetPostsBatchForIndexing(startTime int64, limit int) StoreChannel
+	GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) StoreChannel
 	PermanentDeleteBatch(endTime int64, limit int64) StoreChannel
+	GetOldest() StoreChannel
 }
 
 type UserStore interface {
diff --git a/store/storetest/post_store.go b/store/storetest/post_store.go
index 3460c4d05..b288cde34 100644
--- a/store/storetest/post_store.go
+++ b/store/storetest/post_store.go
@@ -12,6 +12,7 @@ import (
 	"github.com/mattermost/mattermost-server/model"
 	"github.com/mattermost/mattermost-server/store"
 	"github.com/mattermost/mattermost-server/utils"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestPostStore(t *testing.T, ss store.Store) {
@@ -1614,7 +1615,7 @@ func testPostStoreGetPostsBatchForIndexing(t *testing.T, ss store.Store) {
 	o3.Message = "zz" + model.NewId() + "QQQQQQQQQQ"
 	o3 = (<-ss.Post().Save(o3)).Data.(*model.Post)
 
-	if r := store.Must(ss.Post().GetPostsBatchForIndexing(o1.CreateAt, 100)).([]*model.PostForIndexing); len(r) != 3 {
+	if r := store.Must(ss.Post().GetPostsBatchForIndexing(o1.CreateAt, model.GetMillis()+100000, 100)).([]*model.PostForIndexing); len(r) != 3 {
 		t.Fatalf("Expected 3 posts in results. Got %v", len(r))
 	} else {
 		for _, p := range r {
@@ -1682,3 +1683,31 @@ func testPostStorePermanentDeleteBatch(t *testing.T, ss store.Store) {
 		t.Fatalf("Should have found post 3 after purge")
 	}
 }
+
+func testPostStoreGetOldest(t *testing.T, ss store.Store) {
+	o0 := &model.Post{}
+	o0.ChannelId = model.NewId()
+	o0.UserId = model.NewId()
+	o0.Message = "zz" + model.NewId() + "b"
+	o0.CreateAt = 3
+	o0 = (<-ss.Post().Save(o0)).Data.(*model.Post)
+
+	o1 := &model.Post{}
+	o1.ChannelId = o0.Id
+	o1.UserId = model.NewId()
+	o1.Message = "zz" + model.NewId() + "b"
+	o1.CreateAt = 2
+	o1 = (<-ss.Post().Save(o1)).Data.(*model.Post)
+
+	o2 := &model.Post{}
+	o2.Id = model.NewId()
+	o2.ChannelId = o1.ChannelId
+	o2.UserId = model.NewId()
+	o2.Message = "zz" + model.NewId() + "b"
+	o2.CreateAt = 1
+	o2 = (<-ss.Post().Save(o2)).Data.(*model.Post)
+
+	r1 := (<-ss.Post().GetOldest()).Data.(*model.Post)
+
+	assert.EqualValues(t, o2.Id, r1.Id)
+}