From f8a31194263ac0dfb50e15006267a7d6a43f33ff Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Thu, 31 Mar 2022 10:46:01 +0530 Subject: [PATCH] MM-41260: Revamp ES/Bleve batching logic (#19841) The older method used to reply completely on timestamps to take batches of items in a timestamp range and then just incrementing the timestamp. This led to handling edge-cases such as more items than the batch count, all having the same timestamp. Additionally, relying on timestamp as the page cursor meant that indexing was not very efficient if you had several items spread out across large spans of time. To get away from all of that we use a proper cursor-based approach consisting of createAt+Id. With this, we move completely to a constant page size where we can fetch a given number of objects irrespective of when they were created. This makes indexing much more faster and efficient. https://mattermost.atlassian.net/browse/MM-41260 ```release-note Elasticsearch and Bleve indexing have been revamped to be much more efficient and faster. The config parameter BulkIndexingTimeWindowSeconds for both elasticsearch and bleve have been removed. A new config parameter called BatchSize has been introduced instead. This parameter controls the number of objects that can be indexed in a single batch. This makes things more efficient and maintains a constant workload. ``` --- .../mysql/000080_posts_createat_id.down.sql | 14 + .../mysql/000080_posts_createat_id.up.sql | 14 + .../000080_posts_createat_id.down.sql | 1 + .../postgres/000080_posts_createat_id.up.sql | 1 + i18n/en.json | 8 +- model/config.go | 58 ++-- model/config_test.go | 7 + model/utils_test.go | 7 + .../bleveengine/indexer/indexing_job.go | 270 +++++++++--------- services/telemetry/telemetry.go | 46 +-- store/opentracinglayer/opentracinglayer.go | 16 +- store/retrylayer/retrylayer.go | 16 +- store/sqlstore/channel_store.go | 12 +- store/sqlstore/file_info_store.go | 14 +- store/sqlstore/post_store.go | 29 +- store/sqlstore/user_store.go | 13 +- store/store.go | 8 +- store/storetest/channel_store.go | 18 +- store/storetest/file_info_store.go | 51 ++-- store/storetest/mocks/ChannelStore.go | 14 +- store/storetest/mocks/FileInfoStore.go | 14 +- store/storetest/mocks/PostStore.go | 14 +- store/storetest/mocks/UserStore.go | 14 +- store/storetest/post_store.go | 38 +-- store/storetest/user_store.go | 45 +-- store/timerlayer/timerlayer.go | 16 +- tests/test-config.json | 2 +- 27 files changed, 384 insertions(+), 376 deletions(-) create mode 100644 db/migrations/mysql/000080_posts_createat_id.down.sql create mode 100644 db/migrations/mysql/000080_posts_createat_id.up.sql create mode 100644 db/migrations/postgres/000080_posts_createat_id.down.sql create mode 100644 db/migrations/postgres/000080_posts_createat_id.up.sql diff --git a/db/migrations/mysql/000080_posts_createat_id.down.sql b/db/migrations/mysql/000080_posts_createat_id.down.sql new file mode 100644 index 0000000000..214e53c5d3 --- /dev/null +++ b/db/migrations/mysql/000080_posts_createat_id.down.sql @@ -0,0 +1,14 @@ +SET @preparedStatement = (SELECT IF( + ( + SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS + WHERE table_name = 'Posts' + AND table_schema = DATABASE() + AND index_name = 'idx_posts_create_at_id' + ) > 0, + 'DROP INDEX idx_posts_create_at_id on Posts;', + 'SELECT 1;' +)); + +PREPARE removeIndexIfExists FROM @preparedStatement; +EXECUTE removeIndexIfExists; +DEALLOCATE PREPARE removeIndexIfExists; diff --git a/db/migrations/mysql/000080_posts_createat_id.up.sql b/db/migrations/mysql/000080_posts_createat_id.up.sql new file mode 100644 index 0000000000..3548dd1b7b --- /dev/null +++ b/db/migrations/mysql/000080_posts_createat_id.up.sql @@ -0,0 +1,14 @@ +SET @preparedStatement = (SELECT IF( + ( + SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS + WHERE table_name = 'Posts' + AND table_schema = DATABASE() + AND index_name = 'idx_posts_create_at_id' + ) > 0, + 'SELECT 1;', + 'CREATE INDEX idx_posts_create_at_id on Posts(CreateAt, Id) LOCK=NONE;' +)); + +PREPARE createIndexIfNotExists FROM @preparedStatement; +EXECUTE createIndexIfNotExists; +DEALLOCATE PREPARE createIndexIfNotExists; diff --git a/db/migrations/postgres/000080_posts_createat_id.down.sql b/db/migrations/postgres/000080_posts_createat_id.down.sql new file mode 100644 index 0000000000..51c4bc2e60 --- /dev/null +++ b/db/migrations/postgres/000080_posts_createat_id.down.sql @@ -0,0 +1 @@ +DROP INDEX IF exists idx_posts_create_at_id; \ No newline at end of file diff --git a/db/migrations/postgres/000080_posts_createat_id.up.sql b/db/migrations/postgres/000080_posts_createat_id.up.sql new file mode 100644 index 0000000000..e671689735 --- /dev/null +++ b/db/migrations/postgres/000080_posts_createat_id.up.sql @@ -0,0 +1 @@ +CREATE INDEX IF NOT EXISTS idx_posts_create_at_id on posts(createat, id); \ No newline at end of file diff --git a/i18n/en.json b/i18n/en.json index 3ea21a708a..9207629ffc 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -8044,8 +8044,8 @@ "translation": "Invalid RemoteImageProxyURL for atmos/camo. Must be set to your shared key." }, { - "id": "model.config.is_valid.bleve_search.bulk_indexing_time_window_seconds.app_error", - "translation": "Bleve Bulk Indexing Time Window must be at least 1 second." + "id": "model.config.is_valid.bleve_search.bulk_indexing_batch_size.app_error", + "translation": "Bleve Bulk Indexing Batch Size must be at least {{.BatchSize}}." }, { "id": "model.config.is_valid.bleve_search.enable_autocomplete.app_error", @@ -8096,8 +8096,8 @@ "translation": "Elasticsearch AggregatePostsAfterDays setting must be a number greater than or equal to 1." }, { - "id": "model.config.is_valid.elastic_search.bulk_indexing_time_window_seconds.app_error", - "translation": "Elasticsearch Bulk Indexing Time Window must be at least 1 second." + "id": "model.config.is_valid.elastic_search.bulk_indexing_batch_size.app_error", + "translation": "Elasticsearch Bulk Indexing Batch Size must be at least {{.BatchSize}}." }, { "id": "model.config.is_valid.elastic_search.connection_url.app_error", diff --git a/model/config.go b/model/config.go index 32e7f07d60..49fb02e6db 100644 --- a/model/config.go +++ b/model/config.go @@ -184,24 +184,24 @@ const ( TeamSettingsDefaultTeamText = "default" - ElasticsearchSettingsDefaultConnectionURL = "http://localhost:9200" - ElasticsearchSettingsDefaultUsername = "elastic" - ElasticsearchSettingsDefaultPassword = "changeme" - ElasticsearchSettingsDefaultPostIndexReplicas = 1 - ElasticsearchSettingsDefaultPostIndexShards = 1 - ElasticsearchSettingsDefaultChannelIndexReplicas = 1 - ElasticsearchSettingsDefaultChannelIndexShards = 1 - ElasticsearchSettingsDefaultUserIndexReplicas = 1 - ElasticsearchSettingsDefaultUserIndexShards = 1 - ElasticsearchSettingsDefaultAggregatePostsAfterDays = 365 - ElasticsearchSettingsDefaultPostsAggregatorJobStartTime = "03:00" - ElasticsearchSettingsDefaultIndexPrefix = "" - ElasticsearchSettingsDefaultLiveIndexingBatchSize = 1 - ElasticsearchSettingsDefaultBulkIndexingTimeWindowSeconds = 3600 - ElasticsearchSettingsDefaultRequestTimeoutSeconds = 30 + ElasticsearchSettingsDefaultConnectionURL = "http://localhost:9200" + ElasticsearchSettingsDefaultUsername = "elastic" + ElasticsearchSettingsDefaultPassword = "changeme" + ElasticsearchSettingsDefaultPostIndexReplicas = 1 + ElasticsearchSettingsDefaultPostIndexShards = 1 + ElasticsearchSettingsDefaultChannelIndexReplicas = 1 + ElasticsearchSettingsDefaultChannelIndexShards = 1 + ElasticsearchSettingsDefaultUserIndexReplicas = 1 + ElasticsearchSettingsDefaultUserIndexShards = 1 + ElasticsearchSettingsDefaultAggregatePostsAfterDays = 365 + ElasticsearchSettingsDefaultPostsAggregatorJobStartTime = "03:00" + ElasticsearchSettingsDefaultIndexPrefix = "" + ElasticsearchSettingsDefaultLiveIndexingBatchSize = 1 + ElasticsearchSettingsDefaultRequestTimeoutSeconds = 30 + ElasticsearchSettingsDefaultBatchSize = 10000 - BleveSettingsDefaultIndexDir = "" - BleveSettingsDefaultBulkIndexingTimeWindowSeconds = 3600 + BleveSettingsDefaultIndexDir = "" + BleveSettingsDefaultBatchSize = 10000 DataRetentionSettingsDefaultMessageRetentionDays = 365 DataRetentionSettingsDefaultFileRetentionDays = 365 @@ -2480,7 +2480,8 @@ type ElasticsearchSettings struct { PostsAggregatorJobStartTime *string `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` // telemetry: none IndexPrefix *string `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` LiveIndexingBatchSize *int `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` - BulkIndexingTimeWindowSeconds *int `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` + BulkIndexingTimeWindowSeconds *int `json:",omitempty"` // telemetry: none + BatchSize *int `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` RequestTimeoutSeconds *int `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` SkipTLSVerification *bool `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` Trace *string `access:"environment_elasticsearch,write_restrictable,cloud_restrictable"` @@ -2555,8 +2556,8 @@ func (s *ElasticsearchSettings) SetDefaults() { s.LiveIndexingBatchSize = NewInt(ElasticsearchSettingsDefaultLiveIndexingBatchSize) } - if s.BulkIndexingTimeWindowSeconds == nil { - s.BulkIndexingTimeWindowSeconds = NewInt(ElasticsearchSettingsDefaultBulkIndexingTimeWindowSeconds) + if s.BatchSize == nil { + s.BatchSize = NewInt(ElasticsearchSettingsDefaultBatchSize) } if s.RequestTimeoutSeconds == nil { @@ -2577,7 +2578,8 @@ type BleveSettings struct { EnableIndexing *bool `access:"experimental_bleve"` EnableSearching *bool `access:"experimental_bleve"` EnableAutocomplete *bool `access:"experimental_bleve"` - BulkIndexingTimeWindowSeconds *int `access:"experimental_bleve"` + BulkIndexingTimeWindowSeconds *int `json:",omitempty"` // telemetry: none + BatchSize *int `access:"experimental_bleve"` } func (bs *BleveSettings) SetDefaults() { @@ -2597,8 +2599,8 @@ func (bs *BleveSettings) SetDefaults() { bs.EnableAutocomplete = NewBool(false) } - if bs.BulkIndexingTimeWindowSeconds == nil { - bs.BulkIndexingTimeWindowSeconds = NewInt(BleveSettingsDefaultBulkIndexingTimeWindowSeconds) + if bs.BatchSize == nil { + bs.BatchSize = NewInt(BleveSettingsDefaultBatchSize) } } @@ -3632,8 +3634,9 @@ func (s *ElasticsearchSettings) isValid() *AppError { return NewAppError("Config.IsValid", "model.config.is_valid.elastic_search.live_indexing_batch_size.app_error", nil, "", http.StatusBadRequest) } - if *s.BulkIndexingTimeWindowSeconds < 1 { - return NewAppError("Config.IsValid", "model.config.is_valid.elastic_search.bulk_indexing_time_window_seconds.app_error", nil, "", http.StatusBadRequest) + minBatchSize := 1 + if *s.BatchSize < minBatchSize { + return NewAppError("Config.IsValid", "model.config.is_valid.elastic_search.bulk_indexing_batch_size.app_error", map[string]interface{}{"BatchSize": minBatchSize}, "", http.StatusBadRequest) } if *s.RequestTimeoutSeconds < 1 { @@ -3656,8 +3659,9 @@ func (bs *BleveSettings) isValid() *AppError { return NewAppError("Config.IsValid", "model.config.is_valid.bleve_search.enable_autocomplete.app_error", nil, "", http.StatusBadRequest) } } - if *bs.BulkIndexingTimeWindowSeconds < 1 { - return NewAppError("Config.IsValid", "model.config.is_valid.bleve_search.bulk_indexing_time_window_seconds.app_error", nil, "", http.StatusBadRequest) + minBatchSize := 1 + if *bs.BatchSize < minBatchSize { + return NewAppError("Config.IsValid", "model.config.is_valid.bleve_search.bulk_indexing_batch_size.app_error", map[string]interface{}{"BatchSize": minBatchSize}, "", http.StatusBadRequest) } return nil diff --git a/model/config_test.go b/model/config_test.go index e8c15abe2f..9bb16d94c4 100644 --- a/model/config_test.go +++ b/model/config_test.go @@ -31,6 +31,13 @@ func TestConfigDefaults(t *testing.T) { var recursivelyUninitialize func(*Config, string, reflect.Value) recursivelyUninitialize = func(config *Config, name string, v reflect.Value) { if v.Type().Kind() == reflect.Ptr { + // Ignoring these 2 settings. + // TODO: remove them completely in v8.0. + if name == "config.BleveSettings.BulkIndexingTimeWindowSeconds" || + name == "config.ElasticsearchSettings.BulkIndexingTimeWindowSeconds" { + return + } + // Set every pointer we find in the tree to nil v.Set(reflect.Zero(v.Type())) require.True(t, v.IsNil()) diff --git a/model/utils_test.go b/model/utils_test.go index b30763bd4a..a584a0f6f7 100644 --- a/model/utils_test.go +++ b/model/utils_test.go @@ -778,6 +778,13 @@ func checkNowhereNil(t *testing.T, name string, value interface{}) bool { v := reflect.ValueOf(value) switch v.Type().Kind() { case reflect.Ptr: + // Ignoring these 2 settings. + // TODO: remove them completely in v8.0. + if name == "config.BleveSettings.BulkIndexingTimeWindowSeconds" || + name == "config.ElasticsearchSettings.BulkIndexingTimeWindowSeconds" { + return true + } + if v.IsNil() { t.Logf("%s was nil", name) return false diff --git a/services/searchengine/bleveengine/indexer/indexing_job.go b/services/searchengine/bleveengine/indexer/indexing_job.go index db52e7eaeb..0a4851577f 100644 --- a/services/searchengine/bleveengine/indexer/indexing_job.go +++ b/services/searchengine/bleveengine/indexer/indexing_job.go @@ -17,12 +17,12 @@ import ( ) const ( - BatchSize = 1000 - TimeBetweenBatches = 100 - EstimatedPostCount = 10000000 - EstimatedFilesCount = 100000 - EstimatedChannelCount = 100000 - EstimatedUserCount = 10000 + timeBetweenBatches = 100 * time.Millisecond + + estimatedPostCount = 10000000 + estimatedFilesCount = 100000 + estimatedChannelCount = 100000 + estimatedUserCount = 10000 ) type BleveIndexerWorker struct { @@ -50,22 +50,30 @@ func MakeWorker(jobServer *jobs.JobServer, engine *bleveengine.BleveEngine) mode } type IndexingProgress struct { - Now time.Time - StartAtTime int64 - EndAtTime int64 - LastEntityTime int64 - TotalPostsCount int64 - DonePostsCount int64 - DonePosts bool - TotalFilesCount int64 - DoneFilesCount int64 - DoneFiles bool + Now time.Time + StartAtTime int64 + EndAtTime int64 + LastEntityTime int64 + + TotalPostsCount int64 + DonePostsCount int64 + DonePosts bool + LastPostID string + + TotalFilesCount int64 + DoneFilesCount int64 + DoneFiles bool + LastFileID string + TotalChannelsCount int64 DoneChannelsCount int64 DoneChannels bool - TotalUsersCount int64 - DoneUsersCount int64 - DoneUsers bool + LastChannelID string + + TotalUsersCount int64 + DoneUsersCount int64 + DoneUsers bool + LastUserID string } func (ip *IndexingProgress) CurrentProgress() int64 { @@ -160,7 +168,6 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) { return } progress.StartAtTime = startInt - progress.LastEntityTime = progress.StartAtTime } else { // Set start time to oldest entity in the database. // A user or a channel may be created before any post. @@ -174,8 +181,8 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) { return } progress.StartAtTime = oldestEntityCreationTime - progress.LastEntityTime = progress.StartAtTime } + progress.LastEntityTime = progress.StartAtTime if endString, ok := job.Data["end_time"]; ok { endInt, err := strconv.ParseInt(endString, 10, 64) @@ -190,27 +197,43 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) { progress.EndAtTime = endInt } + if id, ok := job.Data["start_post_id"]; ok { + progress.LastPostID = id + } + if id, ok := job.Data["start_channel_id"]; ok { + progress.LastChannelID = id + } + if id, ok := job.Data["start_user_id"]; ok { + progress.LastUserID = id + } + if id, ok := job.Data["start_file_id"]; ok { + progress.LastFileID = id + } + // Counting all posts may fail or timeout when the posts table is large. If this happens, log a warning, but carry // on with the indexing job anyway. The only issue is that the progress % reporting will be inaccurate. if count, err := worker.jobServer.Store.Post().AnalyticsPostCount("", false, false); err != nil { mlog.Warn("Worker: Failed to fetch total post count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) - progress.TotalPostsCount = EstimatedPostCount + progress.TotalPostsCount = estimatedPostCount } else { progress.TotalPostsCount = count } // Same possible fail as above can happen when counting channels - if count, err := worker.jobServer.Store.Channel().AnalyticsTypeCount("", model.ChannelTypeOpen); err != nil { + if count, err := worker.jobServer.Store.Channel().AnalyticsTypeCount("", ""); err != nil { mlog.Warn("Worker: Failed to fetch total channel count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) - progress.TotalChannelsCount = EstimatedChannelCount + progress.TotalChannelsCount = estimatedChannelCount } else { progress.TotalChannelsCount = count } // Same possible fail as above can happen when counting users - if count, err := worker.jobServer.Store.User().Count(model.UserCountOptions{}); err != nil { + if count, err := worker.jobServer.Store.User().Count(model.UserCountOptions{ + IncludeBotAccounts: true, // This actually doesn't join with the bots table + // since ExcludeRegularUsers is set to false + }); err != nil { mlog.Warn("Worker: Failed to fetch total user count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) - progress.TotalUsersCount = EstimatedUserCount + progress.TotalUsersCount = estimatedUserCount } else { progress.TotalUsersCount = count } @@ -219,7 +242,7 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) { // on with the indexing job anyway. The only issue is that the progress % reporting will be inaccurate. if count, err := worker.jobServer.Store.FileInfo().CountAll(); err != nil { mlog.Warn("Worker: Failed to fetch total file info count for job. An estimated value will be used for progress reporting.", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) - progress.TotalFilesCount = EstimatedFilesCount + progress.TotalFilesCount = estimatedFilesCount } else { progress.TotalFilesCount = count } @@ -246,7 +269,7 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) { } return - case <-time.After(TimeBetweenBatches * time.Millisecond): + case <-time.After(timeBetweenBatches): var err *model.AppError if progress, err = worker.IndexBatch(progress); err != nil { mlog.Error("Worker: Failed to index batch for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) @@ -256,6 +279,19 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) { return } + // Storing the batch progress in metadata. + if job.Data == nil { + job.Data = make(model.StringMap) + } + + job.Data["start_time"] = strconv.FormatInt(progress.LastEntityTime, 10) + job.Data["start_post_id"] = progress.LastPostID + job.Data["start_channel_id"] = progress.LastChannelID + job.Data["start_user_id"] = progress.LastUserID + job.Data["start_file_id"] = progress.LastFileID + job.Data["original_start_time"] = strconv.FormatInt(progress.StartAtTime, 10) + job.Data["end_time"] = strconv.FormatInt(progress.EndAtTime, 10) + if err := worker.jobServer.SetJobProgress(job, progress.CurrentProgress()); err != nil { mlog.Error("Worker: Failed to set progress for job", mlog.String("workername", worker.name), mlog.String("job_id", job.Id), mlog.Err(err)) if err2 := worker.jobServer.SetJobError(job, err); err2 != nil { @@ -295,14 +331,12 @@ func (worker *BleveIndexerWorker) IndexBatch(progress IndexingProgress) (Indexin } func (worker *BleveIndexerWorker) IndexPostsBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) { - endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000) - var posts []*model.PostForIndexing tries := 0 for posts == nil { var err error - posts, err = worker.jobServer.Store.Post().GetPostsBatchForIndexing(progress.LastEntityTime, endTime, BatchSize) + posts, err = worker.jobServer.Store.Post().GetPostsBatchForIndexing(progress.LastEntityTime, progress.LastPostID, *worker.jobServer.Config().BleveSettings.BatchSize) if err != nil { if tries >= 10 { return progress, model.NewAppError("IndexPostsBatch", "app.post.get_posts_batch_for_indexing.get.app_error", nil, err.Error(), http.StatusInternalServerError) @@ -316,40 +350,34 @@ func (worker *BleveIndexerWorker) IndexPostsBatch(progress IndexingProgress) (In tries++ } - newLastMessageTime, err := worker.BulkIndexPosts(posts, progress) + // Handle zero messages. + if len(posts) == 0 { + progress.DonePosts = true + progress.LastEntityTime = progress.StartAtTime + return progress, nil + } + + lastPost, err := worker.BulkIndexPosts(posts, progress) if err != nil { return progress, err } - // Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this - // case, set the "newLastMessageTime" to the endTime so we don't get stuck running the same query in a loop. - if len(posts) < BatchSize { - newLastMessageTime = endTime - } - - // When to Stop: we index either until we pass a batch of messages where the last - // message is created at or after the specified end time when setting up the batch - // index, or until two consecutive full batches have the same end time of their final - // messages. This second case is safe as long as the assumption that the database - // cannot contain more messages with the same CreateAt time than the batch size holds. - if progress.EndAtTime <= newLastMessageTime { - progress.DonePosts = true - progress.LastEntityTime = progress.StartAtTime - } else if progress.LastEntityTime == newLastMessageTime && len(posts) == BatchSize { - mlog.Warn("More posts with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastMessageTime), mlog.Int("Batch Size", BatchSize)) + // Our exit condition is when the last post's createAt reaches the initial endAtTime + // set during job creation. + if progress.EndAtTime <= lastPost.CreateAt { progress.DonePosts = true progress.LastEntityTime = progress.StartAtTime } else { - progress.LastEntityTime = newLastMessageTime + progress.LastEntityTime = lastPost.CreateAt } + progress.LastPostID = lastPost.Id progress.DonePostsCount += int64(len(posts)) return progress, nil } -func (worker *BleveIndexerWorker) BulkIndexPosts(posts []*model.PostForIndexing, progress IndexingProgress) (int64, *model.AppError) { - lastCreateAt := int64(0) +func (worker *BleveIndexerWorker) BulkIndexPosts(posts []*model.PostForIndexing, progress IndexingProgress) (*model.Post, *model.AppError) { batch := worker.engine.PostIndex.NewBatch() for _, post := range posts { @@ -359,28 +387,24 @@ func (worker *BleveIndexerWorker) BulkIndexPosts(posts []*model.PostForIndexing, } else { batch.Delete(post.Id) } - - lastCreateAt = post.CreateAt } worker.engine.Mutex.RLock() defer worker.engine.Mutex.RUnlock() if err := worker.engine.PostIndex.Batch(batch); err != nil { - return 0, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_posts.batch_error", nil, err.Error(), http.StatusInternalServerError) + return nil, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_posts.batch_error", nil, err.Error(), http.StatusInternalServerError) } - return lastCreateAt, nil + return &posts[len(posts)-1].Post, nil } func (worker *BleveIndexerWorker) IndexFilesBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) { - endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000) - var files []*model.FileForIndexing tries := 0 for files == nil { var err error - files, err = worker.jobServer.Store.FileInfo().GetFilesBatchForIndexing(progress.LastEntityTime, endTime, BatchSize) + files, err = worker.jobServer.Store.FileInfo().GetFilesBatchForIndexing(progress.LastEntityTime, progress.LastFileID, *worker.jobServer.Config().BleveSettings.BatchSize) if err != nil { if tries >= 10 { return progress, model.NewAppError("IndexFilesBatch", "app.post.get_files_batch_for_indexing.get.app_error", nil, err.Error(), http.StatusInternalServerError) @@ -394,40 +418,33 @@ func (worker *BleveIndexerWorker) IndexFilesBatch(progress IndexingProgress) (In tries++ } - newLastFileTime, err := worker.BulkIndexFiles(files, progress) + if len(files) == 0 { + progress.DoneFiles = true + progress.LastEntityTime = progress.StartAtTime + return progress, nil + } + + lastFile, err := worker.BulkIndexFiles(files, progress) if err != nil { return progress, err } - // Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this - // case, set the "newLastFileTime" to the endTime so we don't get stuck running the same query in a loop. - if len(files) < BatchSize { - newLastFileTime = endTime - } - - // When to Stop: we index either until we pass a batch of messages where the last - // message is created at or after the specified end time when setting up the batch - // index, or until two consecutive full batches have the same end time of their final - // messages. This second case is safe as long as the assumption that the database - // cannot contain more messages with the same CreateAt time than the batch size holds. - if progress.EndAtTime <= newLastFileTime { - progress.DoneFiles = true - progress.LastEntityTime = progress.StartAtTime - } else if progress.LastEntityTime == newLastFileTime && len(files) == BatchSize { - mlog.Warn("More files with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastFileTime), mlog.Int("Batch Size", BatchSize)) + // Our exit condition is when the last file's createAt reaches the initial endAtTime + // set during job creation. + if progress.EndAtTime <= lastFile.CreateAt { progress.DoneFiles = true progress.LastEntityTime = progress.StartAtTime } else { - progress.LastEntityTime = newLastFileTime + progress.LastEntityTime = lastFile.CreateAt } + progress.LastFileID = lastFile.Id progress.DoneFilesCount += int64(len(files)) return progress, nil } -func (worker *BleveIndexerWorker) BulkIndexFiles(files []*model.FileForIndexing, progress IndexingProgress) (int64, *model.AppError) { - lastCreateAt := int64(0) +func (worker *BleveIndexerWorker) BulkIndexFiles(files []*model.FileForIndexing, progress IndexingProgress) (*model.FileInfo, *model.AppError) { batch := worker.engine.FileIndex.NewBatch() for _, file := range files { @@ -437,28 +454,24 @@ func (worker *BleveIndexerWorker) BulkIndexFiles(files []*model.FileForIndexing, } else { batch.Delete(file.Id) } - - lastCreateAt = file.CreateAt } worker.engine.Mutex.RLock() defer worker.engine.Mutex.RUnlock() if err := worker.engine.FileIndex.Batch(batch); err != nil { - return 0, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_files.batch_error", nil, err.Error(), http.StatusInternalServerError) + return nil, model.NewAppError("BleveIndexerWorker.BulkIndexPosts", "bleveengine.indexer.do_job.bulk_index_files.batch_error", nil, err.Error(), http.StatusInternalServerError) } - return lastCreateAt, nil + return &files[len(files)-1].FileInfo, nil } func (worker *BleveIndexerWorker) IndexChannelsBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) { - endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000) - var channels []*model.Channel tries := 0 for channels == nil { var nErr error - channels, nErr = worker.jobServer.Store.Channel().GetChannelsBatchForIndexing(progress.LastEntityTime, endTime, BatchSize) + channels, nErr = worker.jobServer.Store.Channel().GetChannelsBatchForIndexing(progress.LastEntityTime, progress.LastChannelID, *worker.jobServer.Config().BleveSettings.BatchSize) if nErr != nil { if tries >= 10 { return progress, model.NewAppError("BleveIndexerWorker.IndexChannelsBatch", "app.channel.get_channels_batch_for_indexing.get.app_error", nil, nErr.Error(), http.StatusInternalServerError) @@ -472,40 +485,33 @@ func (worker *BleveIndexerWorker) IndexChannelsBatch(progress IndexingProgress) tries++ } - newLastChannelTime, err := worker.BulkIndexChannels(channels, progress) + if len(channels) == 0 { + progress.DoneChannels = true + progress.LastEntityTime = progress.StartAtTime + return progress, nil + } + + lastChannel, err := worker.BulkIndexChannels(channels, progress) if err != nil { return progress, err } - // Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this - // case, set the "newLastChannelTime" to the endTime so we don't get stuck running the same query in a loop. - if len(channels) < BatchSize { - newLastChannelTime = endTime - } - - // When to Stop: we index either until we pass a batch of channels where the last - // channel is created at or after the specified end time when setting up the batch - // index, or until two consecutive full batches have the same end time of their final - // channels. This second case is safe as long as the assumption that the database - // cannot contain more channels with the same CreateAt time than the batch size holds. - if progress.EndAtTime <= newLastChannelTime { - progress.DoneChannels = true - progress.LastEntityTime = progress.StartAtTime - } else if progress.LastEntityTime == newLastChannelTime && len(channels) == BatchSize { - mlog.Warn("More channels with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastChannelTime), mlog.Int("Batch Size", BatchSize)) + // Our exit condition is when the last channel's createAt reaches the initial endAtTime + // set during job creation. + if progress.EndAtTime <= lastChannel.CreateAt { progress.DoneChannels = true progress.LastEntityTime = progress.StartAtTime } else { - progress.LastEntityTime = newLastChannelTime + progress.LastEntityTime = lastChannel.CreateAt } + progress.LastChannelID = lastChannel.Id progress.DoneChannelsCount += int64(len(channels)) return progress, nil } -func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, progress IndexingProgress) (int64, *model.AppError) { - lastCreateAt := int64(0) +func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, progress IndexingProgress) (*model.Channel, *model.AppError) { batch := worker.engine.ChannelIndex.NewBatch() for _, channel := range channels { @@ -515,14 +521,14 @@ func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, p if channel.Type == model.ChannelTypePrivate { userIDs, err = worker.jobServer.Store.Channel().GetAllChannelMembersById(channel.Id) if err != nil { - return 0, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError) + return nil, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError) } } // Get teamMember ids from channelid teamMemberIDs, err := worker.jobServer.Store.Channel().GetTeamMembersForChannel(channel.Id) if err != nil { - return 0, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError) + return nil, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError) } searchChannel := bleveengine.BLVChannelFromChannel(channel, userIDs, teamMemberIDs) @@ -530,27 +536,23 @@ func (worker *BleveIndexerWorker) BulkIndexChannels(channels []*model.Channel, p } else { batch.Delete(channel.Id) } - - lastCreateAt = channel.CreateAt } worker.engine.Mutex.RLock() defer worker.engine.Mutex.RUnlock() if err := worker.engine.ChannelIndex.Batch(batch); err != nil { - return 0, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError) + return nil, model.NewAppError("BleveIndexerWorker.BulkIndexChannels", "bleveengine.indexer.do_job.bulk_index_channels.batch_error", nil, err.Error(), http.StatusInternalServerError) } - return lastCreateAt, nil + return channels[len(channels)-1], nil } func (worker *BleveIndexerWorker) IndexUsersBatch(progress IndexingProgress) (IndexingProgress, *model.AppError) { - endTime := progress.LastEntityTime + int64(*worker.jobServer.Config().BleveSettings.BulkIndexingTimeWindowSeconds*1000) - var users []*model.UserForIndexing tries := 0 for users == nil { - if usersBatch, err := worker.jobServer.Store.User().GetUsersBatchForIndexing(progress.LastEntityTime, endTime, BatchSize); err != nil { + if usersBatch, err := worker.jobServer.Store.User().GetUsersBatchForIndexing(progress.LastEntityTime, progress.LastUserID, *worker.jobServer.Config().BleveSettings.BatchSize); err != nil { if tries >= 10 { return progress, model.NewAppError("IndexUsersBatch", "app.user.get_users_batch_for_indexing.get_users.app_error", nil, err.Error(), http.StatusInternalServerError) } @@ -565,40 +567,32 @@ func (worker *BleveIndexerWorker) IndexUsersBatch(progress IndexingProgress) (In tries++ } - newLastUserTime, err := worker.BulkIndexUsers(users, progress) + if len(users) == 0 { + progress.DoneUsers = true + progress.LastEntityTime = progress.StartAtTime + return progress, nil + } + + lastUser, err := worker.BulkIndexUsers(users, progress) if err != nil { return progress, err } - // Due to the "endTime" parameter in the store query, we might get an incomplete batch before the end. In this - // case, set the "newLastUserTime" to the endTime so we don't get stuck running the same query in a loop. - if len(users) < BatchSize { - newLastUserTime = endTime - } - - // When to Stop: we index either until we pass a batch of users where the last - // user is created at or after the specified end time when setting up the batch - // index, or until two consecutive full batches have the same end time of their final - // users. This second case is safe as long as the assumption that the database - // cannot contain more users with the same CreateAt time than the batch size holds. - if progress.EndAtTime <= newLastUserTime { - progress.DoneUsers = true - progress.LastEntityTime = progress.StartAtTime - } else if progress.LastEntityTime == newLastUserTime && len(users) == BatchSize { - mlog.Warn("More users with the same CreateAt time were detected than the permitted batch size. Aborting indexing job.", mlog.Int64("CreateAt", newLastUserTime), mlog.Int("Batch Size", BatchSize)) + // Our exit condition is when the last user's createAt reaches the initial endAtTime + // set during job creation. + if progress.EndAtTime <= lastUser.CreateAt { progress.DoneUsers = true progress.LastEntityTime = progress.StartAtTime } else { - progress.LastEntityTime = newLastUserTime + progress.LastEntityTime = lastUser.CreateAt } - + progress.LastUserID = lastUser.Id progress.DoneUsersCount += int64(len(users)) return progress, nil } -func (worker *BleveIndexerWorker) BulkIndexUsers(users []*model.UserForIndexing, progress IndexingProgress) (int64, *model.AppError) { - lastCreateAt := int64(0) +func (worker *BleveIndexerWorker) BulkIndexUsers(users []*model.UserForIndexing, progress IndexingProgress) (*model.UserForIndexing, *model.AppError) { batch := worker.engine.UserIndex.NewBatch() for _, user := range users { @@ -608,15 +602,13 @@ func (worker *BleveIndexerWorker) BulkIndexUsers(users []*model.UserForIndexing, } else { batch.Delete(user.Id) } - - lastCreateAt = user.CreateAt } worker.engine.Mutex.RLock() defer worker.engine.Mutex.RUnlock() if err := worker.engine.UserIndex.Batch(batch); err != nil { - return 0, model.NewAppError("BleveIndexerWorker.BulkIndexUsers", "bleveengine.indexer.do_job.bulk_index_users.batch_error", nil, err.Error(), http.StatusInternalServerError) + return nil, model.NewAppError("BleveIndexerWorker.BulkIndexUsers", "bleveengine.indexer.do_job.bulk_index_users.batch_error", nil, err.Error(), http.StatusInternalServerError) } - return lastCreateAt, nil + return users[len(users)-1], nil } diff --git a/services/telemetry/telemetry.go b/services/telemetry/telemetry.go index 2138053dc8..6a2aea5d92 100644 --- a/services/telemetry/telemetry.go +++ b/services/telemetry/telemetry.go @@ -735,25 +735,25 @@ func (ts *TelemetryService) trackConfig() { }) ts.SendTelemetry(TrackConfigElasticsearch, map[string]interface{}{ - "isdefault_connection_url": isDefault(*cfg.ElasticsearchSettings.ConnectionURL, model.ElasticsearchSettingsDefaultConnectionURL), - "isdefault_username": isDefault(*cfg.ElasticsearchSettings.Username, model.ElasticsearchSettingsDefaultUsername), - "isdefault_password": isDefault(*cfg.ElasticsearchSettings.Password, model.ElasticsearchSettingsDefaultPassword), - "enable_indexing": *cfg.ElasticsearchSettings.EnableIndexing, - "enable_searching": *cfg.ElasticsearchSettings.EnableSearching, - "enable_autocomplete": *cfg.ElasticsearchSettings.EnableAutocomplete, - "sniff": *cfg.ElasticsearchSettings.Sniff, - "post_index_replicas": *cfg.ElasticsearchSettings.PostIndexReplicas, - "post_index_shards": *cfg.ElasticsearchSettings.PostIndexShards, - "channel_index_replicas": *cfg.ElasticsearchSettings.ChannelIndexReplicas, - "channel_index_shards": *cfg.ElasticsearchSettings.ChannelIndexShards, - "user_index_replicas": *cfg.ElasticsearchSettings.UserIndexReplicas, - "user_index_shards": *cfg.ElasticsearchSettings.UserIndexShards, - "isdefault_index_prefix": isDefault(*cfg.ElasticsearchSettings.IndexPrefix, model.ElasticsearchSettingsDefaultIndexPrefix), - "live_indexing_batch_size": *cfg.ElasticsearchSettings.LiveIndexingBatchSize, - "bulk_indexing_time_window_seconds": *cfg.ElasticsearchSettings.BulkIndexingTimeWindowSeconds, - "request_timeout_seconds": *cfg.ElasticsearchSettings.RequestTimeoutSeconds, - "skip_tls_verification": *cfg.ElasticsearchSettings.SkipTLSVerification, - "trace": *cfg.ElasticsearchSettings.Trace, + "isdefault_connection_url": isDefault(*cfg.ElasticsearchSettings.ConnectionURL, model.ElasticsearchSettingsDefaultConnectionURL), + "isdefault_username": isDefault(*cfg.ElasticsearchSettings.Username, model.ElasticsearchSettingsDefaultUsername), + "isdefault_password": isDefault(*cfg.ElasticsearchSettings.Password, model.ElasticsearchSettingsDefaultPassword), + "enable_indexing": *cfg.ElasticsearchSettings.EnableIndexing, + "enable_searching": *cfg.ElasticsearchSettings.EnableSearching, + "enable_autocomplete": *cfg.ElasticsearchSettings.EnableAutocomplete, + "sniff": *cfg.ElasticsearchSettings.Sniff, + "post_index_replicas": *cfg.ElasticsearchSettings.PostIndexReplicas, + "post_index_shards": *cfg.ElasticsearchSettings.PostIndexShards, + "channel_index_replicas": *cfg.ElasticsearchSettings.ChannelIndexReplicas, + "channel_index_shards": *cfg.ElasticsearchSettings.ChannelIndexShards, + "user_index_replicas": *cfg.ElasticsearchSettings.UserIndexReplicas, + "user_index_shards": *cfg.ElasticsearchSettings.UserIndexShards, + "isdefault_index_prefix": isDefault(*cfg.ElasticsearchSettings.IndexPrefix, model.ElasticsearchSettingsDefaultIndexPrefix), + "live_indexing_batch_size": *cfg.ElasticsearchSettings.LiveIndexingBatchSize, + "bulk_indexing_batch_size": *cfg.ElasticsearchSettings.BatchSize, + "request_timeout_seconds": *cfg.ElasticsearchSettings.RequestTimeoutSeconds, + "skip_tls_verification": *cfg.ElasticsearchSettings.SkipTLSVerification, + "trace": *cfg.ElasticsearchSettings.Trace, }) ts.trackPluginConfig(cfg, model.PluginSettingsDefaultMarketplaceURL) @@ -804,10 +804,10 @@ func (ts *TelemetryService) trackConfig() { }) ts.SendTelemetry(TrackConfigBleve, map[string]interface{}{ - "enable_indexing": *cfg.BleveSettings.EnableIndexing, - "enable_searching": *cfg.BleveSettings.EnableSearching, - "enable_autocomplete": *cfg.BleveSettings.EnableAutocomplete, - "bulk_indexing_time_window_seconds": *cfg.BleveSettings.BulkIndexingTimeWindowSeconds, + "enable_indexing": *cfg.BleveSettings.EnableIndexing, + "enable_searching": *cfg.BleveSettings.EnableSearching, + "enable_autocomplete": *cfg.BleveSettings.EnableAutocomplete, + "bulk_indexing_batch_size": *cfg.BleveSettings.BatchSize, }) ts.SendTelemetry(TrackConfigExport, map[string]interface{}{ diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 86b2e00c65..5ae97d9564 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -1105,7 +1105,7 @@ func (s *OpenTracingLayerChannelStore) GetChannels(teamID string, userID string, return result, err } -func (s *OpenTracingLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) { +func (s *OpenTracingLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetChannelsBatchForIndexing") s.Root.Store.SetContext(newCtx) @@ -1114,7 +1114,7 @@ func (s *OpenTracingLayerChannelStore) GetChannelsBatchForIndexing(startTime int }() defer span.Finish() - result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, endTime, limit) + result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, startChannelID, limit) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) @@ -3318,7 +3318,7 @@ func (s *OpenTracingLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, return result, err } -func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) { +func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "FileInfoStore.GetFilesBatchForIndexing") s.Root.Store.SetContext(newCtx) @@ -3327,7 +3327,7 @@ func (s *OpenTracingLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64 }() defer span.Finish() - result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit) + result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) @@ -5700,7 +5700,7 @@ func (s *OpenTracingLayerPostStore) GetPostsAfter(options model.GetPostsOptions) return result, err } -func (s *OpenTracingLayerPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) { +func (s *OpenTracingLayerPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetPostsBatchForIndexing") s.Root.Store.SetContext(newCtx) @@ -5709,7 +5709,7 @@ func (s *OpenTracingLayerPostStore) GetPostsBatchForIndexing(startTime int64, en }() defer span.Finish() - result, err := s.PostStore.GetPostsBatchForIndexing(startTime, endTime, limit) + result, err := s.PostStore.GetPostsBatchForIndexing(startTime, startPostID, limit) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) @@ -10610,7 +10610,7 @@ func (s *OpenTracingLayerUserStore) GetUnreadCountForChannel(userID string, chan return result, err } -func (s *OpenTracingLayerUserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) { +func (s *OpenTracingLayerUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "UserStore.GetUsersBatchForIndexing") s.Root.Store.SetContext(newCtx) @@ -10619,7 +10619,7 @@ func (s *OpenTracingLayerUserStore) GetUsersBatchForIndexing(startTime int64, en }() defer span.Finish() - result, err := s.UserStore.GetUsersBatchForIndexing(startTime, endTime, limit) + result, err := s.UserStore.GetUsersBatchForIndexing(startTime, startFileID, limit) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 4925f706ee..8e8154b2ae 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -1234,11 +1234,11 @@ func (s *RetryLayerChannelStore) GetChannels(teamID string, userID string, opts } -func (s *RetryLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) { +func (s *RetryLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) { tries := 0 for { - result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, endTime, limit) + result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, startChannelID, limit) if err == nil { return result, nil } @@ -3715,11 +3715,11 @@ func (s *RetryLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error } -func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) { +func (s *RetryLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { tries := 0 for { - result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit) + result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit) if err == nil { return result, nil } @@ -6448,11 +6448,11 @@ func (s *RetryLayerPostStore) GetPostsAfter(options model.GetPostsOptions) (*mod } -func (s *RetryLayerPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) { +func (s *RetryLayerPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) { tries := 0 for { - result, err := s.PostStore.GetPostsBatchForIndexing(startTime, endTime, limit) + result, err := s.PostStore.GetPostsBatchForIndexing(startTime, startPostID, limit) if err == nil { return result, nil } @@ -12103,11 +12103,11 @@ func (s *RetryLayerUserStore) GetUnreadCountForChannel(userID string, channelID } -func (s *RetryLayerUserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) { +func (s *RetryLayerUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) { tries := 0 for { - result, err := s.UserStore.GetUsersBatchForIndexing(startTime, endTime, limit) + result, err := s.UserStore.GetUsersBatchForIndexing(startTime, startFileID, limit) if err == nil { return result, nil } diff --git a/store/sqlstore/channel_store.go b/store/sqlstore/channel_store.go index 5ee7d58948..a2400a09ad 100644 --- a/store/sqlstore/channel_store.go +++ b/store/sqlstore/channel_store.go @@ -3949,23 +3949,23 @@ func (s SqlChannelStore) GetAllDirectChannelsForExportAfter(limit int, afterId s return directChannelsForExport, nil } -func (s SqlChannelStore) GetChannelsBatchForIndexing(startTime, endTime int64, limit int) ([]*model.Channel, error) { +func (s SqlChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) { query := `SELECT * FROM Channels WHERE - CreateAt >= ? - AND - CreateAt < ? + CreateAt > ? + OR + (CreateAt = ? AND Id > ?) ORDER BY - CreateAt + CreateAt ASC, Id ASC LIMIT ?` channels := []*model.Channel{} - err := s.GetSearchReplicaX().Select(&channels, query, startTime, endTime, limit) + err := s.GetSearchReplicaX().Select(&channels, query, startTime, startTime, startChannelID, limit) if err != nil { return nil, errors.Wrap(err, "failed to find Channels") } diff --git a/store/sqlstore/file_info_store.go b/store/sqlstore/file_info_store.go index 8353571e83..c22584df0f 100644 --- a/store/sqlstore/file_info_store.go +++ b/store/sqlstore/file_info_store.go @@ -688,17 +688,23 @@ func (fs SqlFileInfoStore) CountAll() (int64, error) { return count, nil } -func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime, endTime int64, limit int) ([]*model.FileForIndexing, error) { +func (fs SqlFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { files := []*model.FileForIndexing{} sql, args, _ := fs.getQueryBuilder(). Select(append(fs.queryFields, "Coalesce(p.ChannelId, '') AS ChannelId")...). From("FileInfo"). LeftJoin("Posts AS p ON FileInfo.PostId = p.Id"). - Where(sq.GtOrEq{"FileInfo.CreateAt": startTime}). - Where(sq.Lt{"FileInfo.CreateAt": endTime}). - OrderBy("FileInfo.CreateAt"). + Where(sq.Or{ + sq.Gt{"FileInfo.CreateAt": startTime}, + sq.And{ + sq.Eq{"FileInfo.CreateAt": startTime}, + sq.Gt{"FileInfo.Id": startFileID}, + }, + }). + OrderBy("FileInfo.CreateAt ASC, FileInfo.Id ASC"). Limit(uint64(limit)). ToSql() + err := fs.GetSearchReplicaX().Select(&files, sql, args...) if err != nil { return nil, errors.Wrap(err, "failed to find Files") diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index 624da2b270..bca52e9593 100644 --- a/store/sqlstore/post_store.go +++ b/store/sqlstore/post_store.go @@ -2198,22 +2198,26 @@ func (s *SqlPostStore) GetPostsByIds(postIds []string) ([]*model.Post, error) { return posts, nil } -func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) { +func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) { posts := []*model.PostForIndexing{} - err := s.GetSearchReplicaX().Select(&posts, - `SELECT - PostsQuery.*, Channels.TeamId, ParentPosts.CreateAt ParentCreateAt + table := "Posts" + // We force this index to avoid any chances of index merge intersection. + if s.DriverName() == model.DatabaseDriverMysql { + table += " USE INDEX(idx_posts_create_at_id)" + } + query := `SELECT + PostsQuery.*, Channels.TeamId FROM ( SELECT * FROM - Posts + ` + table + ` WHERE - Posts.CreateAt >= ? - AND - Posts.CreateAt < ? + Posts.CreateAt > ? + OR + (Posts.CreateAt = ? AND Posts.Id > ?) ORDER BY - CreateAt ASC + CreateAt ASC, Id ASC LIMIT ? ) @@ -2223,11 +2227,8 @@ func (s *SqlPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, Channels ON PostsQuery.ChannelId = Channels.Id - LEFT JOIN - Posts ParentPosts - ON - PostsQuery.RootId = ParentPosts.Id`, - startTime, endTime, limit) + ORDER BY CreateAt ASC, Id ASC` + err := s.GetSearchReplicaX().Select(&posts, query, startTime, startTime, startPostID, limit) if err != nil { return nil, errors.Wrap(err, "failed to find Posts") diff --git a/store/sqlstore/user_store.go b/store/sqlstore/user_store.go index 559f8afe5b..46845b3973 100644 --- a/store/sqlstore/user_store.go +++ b/store/sqlstore/user_store.go @@ -1671,12 +1671,17 @@ func (us SqlUserStore) InferSystemInstallDate() (int64, error) { return createAt, nil } -func (us SqlUserStore) GetUsersBatchForIndexing(startTime, endTime int64, limit int) ([]*model.UserForIndexing, error) { +func (us SqlUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) { users := []*model.User{} usersQuery, args, _ := us.usersQuery. - Where(sq.GtOrEq{"u.CreateAt": startTime}). - Where(sq.Lt{"u.CreateAt": endTime}). - OrderBy("u.CreateAt"). + Where(sq.Or{ + sq.Gt{"u.CreateAt": startTime}, + sq.And{ + sq.Eq{"u.CreateAt": startTime}, + sq.Gt{"u.Id": startFileID}, + }, + }). + OrderBy("u.CreateAt ASC, u.Id ASC"). Limit(uint64(limit)). ToSql() err := us.GetSearchReplicaX().Select(&users, usersQuery, args...) diff --git a/store/store.go b/store/store.go index 421496aeac..212d40acb0 100644 --- a/store/store.go +++ b/store/store.go @@ -267,7 +267,7 @@ type ChannelStore interface { GetAllDirectChannelsForExportAfter(limit int, afterID string) ([]*model.DirectChannelForExport, error) GetChannelMembersForExport(userID string, teamID string) ([]*model.ChannelMemberForExport, error) RemoveAllDeactivatedMembers(channelID string) error - GetChannelsBatchForIndexing(startTime, endTime int64, limit int) ([]*model.Channel, error) + GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) UserBelongsToChannels(userID string, channelIds []string) (bool, error) // UpdateMembersRole sets all of the given team members to admins and all of the other members of the team to @@ -351,7 +351,7 @@ type PostStore interface { Overwrite(post *model.Post) (*model.Post, error) OverwriteMultiple(posts []*model.Post) ([]*model.Post, int, error) GetPostsByIds(postIds []string) ([]*model.Post, error) - GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) + GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) PermanentDeleteBatchForRetentionPolicies(now, globalPolicyEndTime, limit int64, cursor model.RetentionPolicyCursor) (int64, model.RetentionPolicyCursor, error) DeleteOrphanedRows(limit int) (deleted int64, err error) PermanentDeleteBatch(endTime int64, limit int64) (int64, error) @@ -430,7 +430,7 @@ type UserStore interface { ClearAllCustomRoleAssignments() error InferSystemInstallDate() (int64, error) GetAllAfter(limit int, afterID string) ([]*model.User, error) - GetUsersBatchForIndexing(startTime, endTime int64, limit int) ([]*model.UserForIndexing, error) + GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) Count(options model.UserCountOptions) (int64, error) GetTeamGroupUsers(teamID string) ([]*model.User, error) GetChannelGroupUsers(channelID string) ([]*model.User, error) @@ -657,7 +657,7 @@ type FileInfoStore interface { SetContent(fileID, content string) error Search(paramsList []*model.SearchParams, userID, teamID string, page, perPage int) (*model.FileInfoList, error) CountAll() (int64, error) - GetFilesBatchForIndexing(startTime, endTime int64, limit int) ([]*model.FileForIndexing, error) + GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) ClearCaches() } diff --git a/store/storetest/channel_store.go b/store/storetest/channel_store.go index e74c59ed00..52d116dd7c 100644 --- a/store/storetest/channel_store.go +++ b/store/storetest/channel_store.go @@ -7601,7 +7601,6 @@ func testChannelStoreGetChannelsBatchForIndexing(t *testing.T, ss store.Store) { require.NoError(t, nErr) time.Sleep(10 * time.Millisecond) - startTime := c2.CreateAt c3 := &model.Channel{} c3.DisplayName = "Channel3" @@ -7633,23 +7632,20 @@ func testChannelStoreGetChannelsBatchForIndexing(t *testing.T, ss store.Store) { _, nErr = ss.Channel().Save(c6, -1) require.NoError(t, nErr) - endTime := c6.CreateAt - // First and last channel should be outside the range - channels, err := ss.Channel().GetChannelsBatchForIndexing(startTime, endTime, 1000) + channels, err := ss.Channel().GetChannelsBatchForIndexing(c1.CreateAt, "", 4) assert.NoError(t, err) - assert.ElementsMatch(t, []*model.Channel{c2, c3, c4, c5}, channels) + assert.Len(t, channels, 4) - // Update the endTime, last channel should be in - endTime = model.GetMillis() - channels, err = ss.Channel().GetChannelsBatchForIndexing(startTime, endTime, 1000) + // From 4th createat+id + channels, err = ss.Channel().GetChannelsBatchForIndexing(channels[3].CreateAt, channels[3].Id, 5) assert.NoError(t, err) - assert.ElementsMatch(t, []*model.Channel{c2, c3, c4, c5, c6}, channels) + assert.Len(t, channels, 2) // Testing the limit - channels, err = ss.Channel().GetChannelsBatchForIndexing(startTime, endTime, 2) + channels, err = ss.Channel().GetChannelsBatchForIndexing(channels[1].CreateAt, channels[1].Id, 1) assert.NoError(t, err) - assert.ElementsMatch(t, []*model.Channel{c2, c3}, channels) + assert.Len(t, channels, 0) } func testGroupSyncedChannelCount(t *testing.T, ss store.Store) { diff --git a/store/storetest/file_info_store.go b/store/storetest/file_info_store.go index 2884dedc5d..15d5e167ef 100644 --- a/store/storetest/file_info_store.go +++ b/store/storetest/file_info_store.go @@ -673,42 +673,23 @@ func testFileInfoStoreGetFilesBatchForIndexing(t *testing.T, ss store.Store) { ss.FileInfo().PermanentDelete(f3.Id) }() - t.Run("get all files", func(t *testing.T) { - r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt, model.GetMillis()+100000, 100) - require.NoError(t, err) - require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r)) - for _, f := range r { - if f.Id == f1.Id { - require.Equal(t, f.ChannelId, o1.ChannelId, "Unexpected channel ID") - require.Equal(t, f.Path, "file1.txt", "Unexpected filename") - } else if f.Id == f2.Id { - require.Equal(t, f.ChannelId, o2.ChannelId, "Unexpected channel ID") - require.Equal(t, f.Path, "file2.txt", "Unexpected filename") - } else if f.Id == f3.Id { - require.Equal(t, f.ChannelId, o3.ChannelId, "Unexpected channel ID") - require.Equal(t, f.Path, "file3.txt", "Unexpected filename") - } else { - require.Fail(t, "unexpected file returned") - } - } - }) + // Getting all + r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 100) + require.NoError(t, err) + require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r)) - t.Run("get files after certain date", func(t *testing.T) { - r, err := ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt+1, model.GetMillis()+100000, 100) - require.NoError(t, err) - require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r)) - for _, f := range r { - if f.Id == f2.Id { - require.Equal(t, f.ChannelId, o2.ChannelId, "Unexpected channel ID") - require.Equal(t, f.Path, "file2.txt", "Unexpected filename") - } else if f.Id == f3.Id { - require.Equal(t, f.ChannelId, o3.ChannelId, "Unexpected channel ID") - require.Equal(t, f.Path, "file3.txt", "Unexpected filename") - } else { - require.Fail(t, "unexpected file returned") - } - } - }) + // Testing pagination + r, err = ss.FileInfo().GetFilesBatchForIndexing(f1.CreateAt-1, "", 2) + require.NoError(t, err) + require.Len(t, r, 2, "Expected 2 posts in results. Got %v", len(r)) + + r, err = ss.FileInfo().GetFilesBatchForIndexing(r[1].CreateAt, r[1].Id, 2) + require.NoError(t, err) + require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r)) + + r, err = ss.FileInfo().GetFilesBatchForIndexing(r[0].CreateAt, r[0].Id, 2) + require.NoError(t, err) + require.Len(t, r, 0, "Expected 0 posts in results. Got %v", len(r)) } func testFileInfoStoreCountAll(t *testing.T, ss store.Store) { diff --git a/store/storetest/mocks/ChannelStore.go b/store/storetest/mocks/ChannelStore.go index 04d9e41828..501d52f120 100644 --- a/store/storetest/mocks/ChannelStore.go +++ b/store/storetest/mocks/ChannelStore.go @@ -697,13 +697,13 @@ func (_m *ChannelStore) GetChannels(teamID string, userID string, opts *model.Ch return r0, r1 } -// GetChannelsBatchForIndexing provides a mock function with given fields: startTime, endTime, limit -func (_m *ChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) { - ret := _m.Called(startTime, endTime, limit) +// GetChannelsBatchForIndexing provides a mock function with given fields: startTime, startChannelID, limit +func (_m *ChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) { + ret := _m.Called(startTime, startChannelID, limit) var r0 []*model.Channel - if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.Channel); ok { - r0 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(0).(func(int64, string, int) []*model.Channel); ok { + r0 = rf(startTime, startChannelID, limit) } else { if ret.Get(0) != nil { r0 = ret.Get(0).([]*model.Channel) @@ -711,8 +711,8 @@ func (_m *ChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int } var r1 error - if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok { - r1 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(1).(func(int64, string, int) error); ok { + r1 = rf(startTime, startChannelID, limit) } else { r1 = ret.Error(1) } diff --git a/store/storetest/mocks/FileInfoStore.go b/store/storetest/mocks/FileInfoStore.go index 1fabed0ea3..863615f0aa 100644 --- a/store/storetest/mocks/FileInfoStore.go +++ b/store/storetest/mocks/FileInfoStore.go @@ -144,13 +144,13 @@ func (_m *FileInfoStore) GetByPath(path string) (*model.FileInfo, error) { return r0, r1 } -// GetFilesBatchForIndexing provides a mock function with given fields: startTime, endTime, limit -func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) { - ret := _m.Called(startTime, endTime, limit) +// GetFilesBatchForIndexing provides a mock function with given fields: startTime, startFileID, limit +func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { + ret := _m.Called(startTime, startFileID, limit) var r0 []*model.FileForIndexing - if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.FileForIndexing); ok { - r0 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(0).(func(int64, string, int) []*model.FileForIndexing); ok { + r0 = rf(startTime, startFileID, limit) } else { if ret.Get(0) != nil { r0 = ret.Get(0).([]*model.FileForIndexing) @@ -158,8 +158,8 @@ func (_m *FileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64 } var r1 error - if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok { - r1 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(1).(func(int64, string, int) error); ok { + r1 = rf(startTime, startFileID, limit) } else { r1 = ret.Error(1) } diff --git a/store/storetest/mocks/PostStore.go b/store/storetest/mocks/PostStore.go index 94e365b4d2..71ecd3b2f1 100644 --- a/store/storetest/mocks/PostStore.go +++ b/store/storetest/mocks/PostStore.go @@ -465,13 +465,13 @@ func (_m *PostStore) GetPostsAfter(options model.GetPostsOptions) (*model.PostLi return r0, r1 } -// GetPostsBatchForIndexing provides a mock function with given fields: startTime, endTime, limit -func (_m *PostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) { - ret := _m.Called(startTime, endTime, limit) +// GetPostsBatchForIndexing provides a mock function with given fields: startTime, startPostID, limit +func (_m *PostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) { + ret := _m.Called(startTime, startPostID, limit) var r0 []*model.PostForIndexing - if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.PostForIndexing); ok { - r0 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(0).(func(int64, string, int) []*model.PostForIndexing); ok { + r0 = rf(startTime, startPostID, limit) } else { if ret.Get(0) != nil { r0 = ret.Get(0).([]*model.PostForIndexing) @@ -479,8 +479,8 @@ func (_m *PostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, li } var r1 error - if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok { - r1 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(1).(func(int64, string, int) error); ok { + r1 = rf(startTime, startPostID, limit) } else { r1 = ret.Error(1) } diff --git a/store/storetest/mocks/UserStore.go b/store/storetest/mocks/UserStore.go index 3aa5d3c7b1..01f96622c3 100644 --- a/store/storetest/mocks/UserStore.go +++ b/store/storetest/mocks/UserStore.go @@ -979,13 +979,13 @@ func (_m *UserStore) GetUnreadCountForChannel(userID string, channelID string) ( return r0, r1 } -// GetUsersBatchForIndexing provides a mock function with given fields: startTime, endTime, limit -func (_m *UserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) { - ret := _m.Called(startTime, endTime, limit) +// GetUsersBatchForIndexing provides a mock function with given fields: startTime, startFileID, limit +func (_m *UserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) { + ret := _m.Called(startTime, startFileID, limit) var r0 []*model.UserForIndexing - if rf, ok := ret.Get(0).(func(int64, int64, int) []*model.UserForIndexing); ok { - r0 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(0).(func(int64, string, int) []*model.UserForIndexing); ok { + r0 = rf(startTime, startFileID, limit) } else { if ret.Get(0) != nil { r0 = ret.Get(0).([]*model.UserForIndexing) @@ -993,8 +993,8 @@ func (_m *UserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, li } var r1 error - if rf, ok := ret.Get(1).(func(int64, int64, int) error); ok { - r1 = rf(startTime, endTime, limit) + if rf, ok := ret.Get(1).(func(int64, string, int) error); ok { + r1 = rf(startTime, startFileID, limit) } else { r1 = ret.Error(1) } diff --git a/store/storetest/post_store.go b/store/storetest/post_store.go index b24f09d9e4..1c476d2e45 100644 --- a/store/storetest/post_store.go +++ b/store/storetest/post_store.go @@ -2953,7 +2953,7 @@ func testPostStoreGetPostsBatchForIndexing(t *testing.T, ss store.Store) { o2.ChannelId = c2.Id o2.UserId = model.NewId() o2.Message = NewTestId() - o2, err = ss.Post().Save(o2) + _, err = ss.Post().Save(o2) require.NoError(t, err) o3 := &model.Post{} @@ -2961,26 +2961,30 @@ func testPostStoreGetPostsBatchForIndexing(t *testing.T, ss store.Store) { o3.UserId = model.NewId() o3.RootId = o1.Id o3.Message = NewTestId() - o3, err = ss.Post().Save(o3) + _, err = ss.Post().Save(o3) require.NoError(t, err) - r, err := ss.Post().GetPostsBatchForIndexing(o1.CreateAt, model.GetMillis()+100000, 100) + // Getting all + r, err := ss.Post().GetPostsBatchForIndexing(o1.CreateAt-1, "", 100) require.NoError(t, err) require.Len(t, r, 3, "Expected 3 posts in results. Got %v", len(r)) - for _, p := range r { - if p.Id == o1.Id { - require.Equal(t, p.TeamId, c1.TeamId, "Unexpected team ID") - require.Nil(t, p.ParentCreateAt, "Unexpected parent create at") - } else if p.Id == o2.Id { - require.Equal(t, p.TeamId, c2.TeamId, "Unexpected team ID") - require.Nil(t, p.ParentCreateAt, "Unexpected parent create at") - } else if p.Id == o3.Id { - require.Equal(t, p.TeamId, c1.TeamId, "Unexpected team ID") - require.Equal(t, *p.ParentCreateAt, o1.CreateAt, "Unexpected parent create at") - } else { - require.Fail(t, "unexpected post returned") - } - } + + // Testing pagination + r, err = ss.Post().GetPostsBatchForIndexing(o1.CreateAt-1, "", 1) + require.NoError(t, err) + require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r)) + + r, err = ss.Post().GetPostsBatchForIndexing(r[0].CreateAt, r[0].Id, 1) + require.NoError(t, err) + require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r)) + + r, err = ss.Post().GetPostsBatchForIndexing(r[0].CreateAt, r[0].Id, 1) + require.NoError(t, err) + require.Len(t, r, 1, "Expected 1 post in results. Got %v", len(r)) + + r, err = ss.Post().GetPostsBatchForIndexing(r[0].CreateAt, r[0].Id, 1) + require.NoError(t, err) + require.Len(t, r, 0, "Expected 0 post in results. Got %v", len(r)) } func testPostStorePermanentDeleteBatch(t *testing.T, ss store.Store) { diff --git a/store/storetest/user_store.go b/store/storetest/user_store.go index 962d63dfe4..8013ec8929 100644 --- a/store/storetest/user_store.go +++ b/store/storetest/user_store.go @@ -4716,7 +4716,6 @@ func testUserStoreGetUsersBatchForIndexing(t *testing.T, ss store.Store) { }) require.NoError(t, err) - startTime := u2.CreateAt time.Sleep(time.Millisecond) u3, err := ss.User().Save(&model.User{ @@ -4744,47 +4743,23 @@ func testUserStoreGetUsersBatchForIndexing(t *testing.T, ss store.Store) { }) require.NoError(t, err) - endTime := u3.CreateAt - - // First and last user should be outside the range - res1List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 100) + // Getting all users + res1List, err := ss.User().GetUsersBatchForIndexing(u1.CreateAt-1, "", 100) require.NoError(t, err) + assert.Len(t, res1List, 3) - assert.Len(t, res1List, 1) - assert.Equal(t, res1List[0].Username, u2.Username) - assert.ElementsMatch(t, res1List[0].TeamsIds, []string{t1.Id}) - assert.ElementsMatch(t, res1List[0].ChannelsIds, []string{cPub1.Id, cPub2.Id}) - - // Update startTime to include first user - startTime = u1.CreateAt - res2List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 100) + // Testing pagination + res2List, err := ss.User().GetUsersBatchForIndexing(u1.CreateAt-1, "", 1) require.NoError(t, err) + assert.Len(t, res2List, 1) + res2List, err = ss.User().GetUsersBatchForIndexing(res2List[0].CreateAt, res2List[0].Id, 2) + require.NoError(t, err) assert.Len(t, res2List, 2) - assert.Equal(t, res2List[0].Username, u1.Username) - assert.Equal(t, res2List[0].ChannelsIds, []string{}) - assert.Equal(t, res2List[0].TeamsIds, []string{}) - assert.Equal(t, res2List[1].Username, u2.Username) - // Update endTime to include last user - endTime = model.GetMillis() - res3List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 100) + res2List, err = ss.User().GetUsersBatchForIndexing(res2List[1].CreateAt, res2List[1].Id, 2) require.NoError(t, err) - - assert.Len(t, res3List, 3) - assert.Equal(t, res3List[0].Username, u1.Username) - assert.Equal(t, res3List[1].Username, u2.Username) - assert.Equal(t, res3List[2].Username, u3.Username) - assert.ElementsMatch(t, res3List[2].TeamsIds, []string{}) - assert.ElementsMatch(t, res3List[2].ChannelsIds, []string{cPub2.Id}) - - // Testing the limit - res4List, err := ss.User().GetUsersBatchForIndexing(startTime, endTime, 2) - require.NoError(t, err) - - assert.Len(t, res4List, 2) - assert.Equal(t, res4List[0].Username, u1.Username) - assert.Equal(t, res4List[1].Username, u2.Username) + assert.Len(t, res2List, 0) } func testUserStoreGetTeamGroupUsers(t *testing.T, ss store.Store) { diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index 4789dc0161..60c4e8378b 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -1029,10 +1029,10 @@ func (s *TimerLayerChannelStore) GetChannels(teamID string, userID string, opts return result, err } -func (s *TimerLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.Channel, error) { +func (s *TimerLayerChannelStore) GetChannelsBatchForIndexing(startTime int64, startChannelID string, limit int) ([]*model.Channel, error) { start := timemodule.Now() - result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, endTime, limit) + result, err := s.ChannelStore.GetChannelsBatchForIndexing(startTime, startChannelID, limit) elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) if s.Root.Metrics != nil { @@ -3036,10 +3036,10 @@ func (s *TimerLayerFileInfoStore) GetByPath(path string) (*model.FileInfo, error return result, err } -func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.FileForIndexing, error) { +func (s *TimerLayerFileInfoStore) GetFilesBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.FileForIndexing, error) { start := timemodule.Now() - result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, endTime, limit) + result, err := s.FileInfoStore.GetFilesBatchForIndexing(startTime, startFileID, limit) elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) if s.Root.Metrics != nil { @@ -5162,10 +5162,10 @@ func (s *TimerLayerPostStore) GetPostsAfter(options model.GetPostsOptions) (*mod return result, err } -func (s *TimerLayerPostStore) GetPostsBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.PostForIndexing, error) { +func (s *TimerLayerPostStore) GetPostsBatchForIndexing(startTime int64, startPostID string, limit int) ([]*model.PostForIndexing, error) { start := timemodule.Now() - result, err := s.PostStore.GetPostsBatchForIndexing(startTime, endTime, limit) + result, err := s.PostStore.GetPostsBatchForIndexing(startTime, startPostID, limit) elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) if s.Root.Metrics != nil { @@ -9557,10 +9557,10 @@ func (s *TimerLayerUserStore) GetUnreadCountForChannel(userID string, channelID return result, err } -func (s *TimerLayerUserStore) GetUsersBatchForIndexing(startTime int64, endTime int64, limit int) ([]*model.UserForIndexing, error) { +func (s *TimerLayerUserStore) GetUsersBatchForIndexing(startTime int64, startFileID string, limit int) ([]*model.UserForIndexing, error) { start := timemodule.Now() - result, err := s.UserStore.GetUsersBatchForIndexing(startTime, endTime, limit) + result, err := s.UserStore.GetUsersBatchForIndexing(startTime, startFileID, limit) elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) if s.Root.Metrics != nil { diff --git a/tests/test-config.json b/tests/test-config.json index e0a27a2c59..3f8cc1a4e7 100644 --- a/tests/test-config.json +++ b/tests/test-config.json @@ -340,7 +340,7 @@ "PostsAggregatorJobStartTime": "03:00", "IndexPrefix": "", "LiveIndexingBatchSize": 1, - "BulkIndexingTimeWindowSeconds": 3600, + "BatchSize": 10000, "RequestTimeoutSeconds": 30 }, "DataRetentionSettings": {