From f5ee7aef525e851eb4fbebc84f9cc021168fd283 Mon Sep 17 00:00:00 2001 From: djamel Date: Thu, 20 Nov 2025 15:52:28 +0100 Subject: [PATCH] Add new parameter to Indicates whether to use minimal indexes for the collections. rework on ReadyTask query to use Mongo query language --- Adaptors/MongoDB/src/Options/MongoDB.cs | 14 +++++--- .../Table/DataModel/TaskDataModelMapping.cs | 32 +++++++++++-------- Adaptors/MongoDB/src/TaskTable.cs | 20 ++++++++---- Compute/PollingAgent/src/appsettings.json | 3 +- Control/Metrics/src/appsettings.json | 3 +- Control/PartitionMetrics/src/appsettings.json | 3 +- Control/Submitter/src/appsettings.json | 3 +- 7 files changed, 50 insertions(+), 28 deletions(-) diff --git a/Adaptors/MongoDB/src/Options/MongoDB.cs b/Adaptors/MongoDB/src/Options/MongoDB.cs index 9dc541575..8510fb84a 100644 --- a/Adaptors/MongoDB/src/Options/MongoDB.cs +++ b/Adaptors/MongoDB/src/Options/MongoDB.cs @@ -1,17 +1,17 @@ // This file is part of the ArmoniK project -// +// // Copyright (C) ANEO, 2021-2025. All rights reserved. -// +// // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published // by the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. -// +// // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY, without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. -// +// // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . @@ -135,4 +135,10 @@ public class MongoDB /// Authentication source for the MongoDB connection. /// public string AuthSource { get; set; } = ""; + + /// + /// Indicates whether to use minimal indexes for the collections. + /// + public bool UseMinimalIndexes { get; set; } = false; + } diff --git a/Adaptors/MongoDB/src/Table/DataModel/TaskDataModelMapping.cs b/Adaptors/MongoDB/src/Table/DataModel/TaskDataModelMapping.cs index 3d34a1c1a..d184e5113 100644 --- a/Adaptors/MongoDB/src/Table/DataModel/TaskDataModelMapping.cs +++ b/Adaptors/MongoDB/src/Table/DataModel/TaskDataModelMapping.cs @@ -1,17 +1,17 @@ // This file is part of the ArmoniK project -// +// // Copyright (C) ANEO, 2021-2025. All rights reserved. -// +// // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published // by the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. -// +// // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY, without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. -// +// // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . @@ -191,25 +191,31 @@ public async Task InitializeIndexesAsync(IClientSessionHandle sessionHandl IMongoCollection collection, Options.MongoDB options) { - var indexModels = new[] + var indexModels = new List> { IndexHelper.CreateHashedIndex(model => model.Status), IndexHelper.CreateHashedIndex(model => model.Options.PartitionId), IndexHelper.CreateHashedIndex(model => model.SessionId), - IndexHelper.CreateHashedIndex(model => model.OwnerPodId), IndexHelper.CreateHashedIndex(model => model.InitialTaskId), IndexHelper.CreateHashedIndex(model => model.CreatedBy), - IndexHelper.CreateAscendingIndex(model => model.CreationDate, - expireAfter: options.DataRetention), - IndexHelper.CreateAscendingIndex(model => model.SubmittedDate), - IndexHelper.CreateAscendingIndex(model => model.StartDate), - IndexHelper.CreateAscendingIndex(model => model.EndDate), - IndexHelper.CreateAscendingIndex(model => model.CreationToEndDuration), - IndexHelper.CreateAscendingIndex(model => model.ProcessingToEndDuration), IndexHelper.CreateCombinedIndex(model => model.Options.PartitionId, model => model.Status), }; + if (options.UseMinimalIndexes == false) + { + indexModels.AddRange([ + IndexHelper.CreateHashedIndex(model => model.OwnerPodId), + IndexHelper.CreateAscendingIndex(model => model.CreationDate, + expireAfter: options.DataRetention), + IndexHelper.CreateAscendingIndex(model => model.SubmittedDate), + IndexHelper.CreateAscendingIndex(model => model.StartDate), + IndexHelper.CreateAscendingIndex(model => model.EndDate), + IndexHelper.CreateAscendingIndex(model => model.CreationToEndDuration), + IndexHelper.CreateAscendingIndex(model => model.ProcessingToEndDuration), + ]); + } + await collection.Indexes.CreateManyAsync(sessionHandle, indexModels) .ConfigureAwait(false); diff --git a/Adaptors/MongoDB/src/TaskTable.cs b/Adaptors/MongoDB/src/TaskTable.cs index 3123c143b..6bba621ef 100644 --- a/Adaptors/MongoDB/src/TaskTable.cs +++ b/Adaptors/MongoDB/src/TaskTable.cs @@ -1,17 +1,17 @@ // This file is part of the ArmoniK project -// +// // Copyright (C) ANEO, 2021-2025. All rights reserved. -// +// // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published // by the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. -// +// // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY, without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. -// +// // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . @@ -409,9 +409,15 @@ await taskCollection.UpdateManyAsync(sessionHandle, cancellationToken: cancellationToken) .ConfigureAwait(false); - var readyTasks = taskCollection.Find(sessionHandle, - data => taskIds.Contains(data.TaskId) && (data.Status == TaskStatus.Creating || data.Status == TaskStatus.Pending) && - data.RemainingDataDependencies == new Dictionary()) + var emptyDictionary = new Dictionary(); + + var filter = Builders.Filter.In(t => t.TaskId, taskIds) + & Builders.Filter.Or(Builders.Filter.Eq(t => t.Status, TaskStatus.Creating), + Builders.Filter.Eq(t => t.Status,TaskStatus.Pending)) + & Builders.Filter.Eq(t => t.RemainingDataDependencies, + emptyDictionary); + + var readyTasks = taskCollection.Find(filter) .Project(selector) .ToAsyncEnumerable(cancellationToken); diff --git a/Compute/PollingAgent/src/appsettings.json b/Compute/PollingAgent/src/appsettings.json index 15e62fe9c..04ed82851 100644 --- a/Compute/PollingAgent/src/appsettings.json +++ b/Compute/PollingAgent/src/appsettings.json @@ -30,7 +30,8 @@ }, "ObjectStorage": { "ChunkSize": "100000" - } + }, + "UseMinimalIndexes" : true }, "Amqp": { "MaxRetries": "10", diff --git a/Control/Metrics/src/appsettings.json b/Control/Metrics/src/appsettings.json index 795dd5b44..2a7708148 100644 --- a/Control/Metrics/src/appsettings.json +++ b/Control/Metrics/src/appsettings.json @@ -28,7 +28,8 @@ }, "ObjectStorage": { "ChunkSize": "100000" - } + }, + "UseMinimalIndexes": true }, "Amqp": { "MaxRetries": "10", diff --git a/Control/PartitionMetrics/src/appsettings.json b/Control/PartitionMetrics/src/appsettings.json index 66ea4e8bb..44f4c0e31 100644 --- a/Control/PartitionMetrics/src/appsettings.json +++ b/Control/PartitionMetrics/src/appsettings.json @@ -33,7 +33,8 @@ "LockRefreshPeriodicity": "00:20:00", "PollPeriodicity": "00:00:50", "LockRefreshExtension": "00:50:00" - } + }, + "UseMinimalIndexes": true }, "Amqp": { "MaxRetries": "10", diff --git a/Control/Submitter/src/appsettings.json b/Control/Submitter/src/appsettings.json index 70131e9b7..d14ce07bb 100644 --- a/Control/Submitter/src/appsettings.json +++ b/Control/Submitter/src/appsettings.json @@ -31,7 +31,8 @@ }, "ObjectStorage": { "ChunkSize": "100000" - } + }, + "UseMinimalIndexes": true }, "Amqp": { "MaxRetries": "10",