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": {