EVA-3639 File stats job

src/main/java/uk/ac/ebi/eva/commons/models/mongo/entity/subdocuments/VariantSourceEntryMongo.java

@@ -189,4 +189,8 @@ public String getFileId() {
     public String[] getAlternates() {
         return alternates;
     }
+
+    public BasicDBObject getAttrs() {
+        return attrs;
+    }
 }

src/main/java/uk/ac/ebi/eva/pipeline/configuration/BeanNames.java

@@ -63,7 +63,8 @@ public class BeanNames {
     public static final String DROP_FILES_BY_STUDY_STEP = "drop-files-by-study-step";
     public static final String LOAD_ANNOTATION_METADATA_STEP = "annotation-metadata-step";
     public static final String ACCESSION_IMPORT_STEP = "accession-import-step";
-    public static final String CALCULATE_AND_LOAD_STATISTICS_STEP = "calculate-load-statistics-step";
+    public static final String VARIANT_STATS_STEP = "variant-stats-step";
+    public static final String FILE_STATS_STEP = "file-stats-step";
 
     public static final String AGGREGATED_VCF_JOB = "aggregated-vcf-job";
     public static final String ANNOTATE_VARIANTS_JOB = "annotate-variants-job";
@@ -73,5 +74,6 @@ public class BeanNames {
     public static final String CALCULATE_STATISTICS_JOB = "calculate-statistics-job";
     public static final String DROP_STUDY_JOB = "drop-study-job";
     public static final String ACCESSION_IMPORT_JOB = "accession-import-job";
-    public static final String CALCULATE_AND_LOAD_STATISTICS_JOB = "calculate-load-statistics-job";
+    public static final String VARIANT_STATS_JOB = "variant-stats-job";
+    public static final String FILE_STATS_JOB = "file-stats-job";
 }

src/main/java/uk/ac/ebi/eva/pipeline/configuration/jobs/CalculateAndLoadStatisticsJobConfiguration.java → src/main/java/uk/ac/ebi/eva/pipeline/configuration/jobs/FileStatsJobConfiguration.java

@@ -27,37 +27,32 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Scope;
-import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.CalculateAndLoadStatisticsStepConfiguration;
+import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.FileStatsStepConfiguration;
 import uk.ac.ebi.eva.pipeline.parameters.NewJobIncrementer;
 
-import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.CALCULATE_AND_LOAD_STATISTICS_JOB;
-import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.CALCULATE_AND_LOAD_STATISTICS_STEP;
+import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.FILE_STATS_JOB;
+import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.FILE_STATS_STEP;
 
-/**
- * Configuration to run a full Statistics job: variantStatsFlow: statsCreate --> statsLoad
- * <p>
- * TODO add a new PopulationStatisticsJobParametersValidator
- */
 @Configuration
 @EnableBatchProcessing
-@Import({CalculateAndLoadStatisticsStepConfiguration.class})
-public class CalculateAndLoadStatisticsJobConfiguration {
+@Import({FileStatsStepConfiguration.class})
+public class FileStatsJobConfiguration {
 
-    private static final Logger logger = LoggerFactory.getLogger(CalculateAndLoadStatisticsJobConfiguration.class);
+    private static final Logger logger = LoggerFactory.getLogger(FileStatsJobConfiguration.class);
 
     @Autowired
-    @Qualifier(CALCULATE_AND_LOAD_STATISTICS_STEP)
-    private Step calculateAndLoadStatisticsStep;
+    @Qualifier(FILE_STATS_STEP)
+    private Step fileStatsStep;
 
-    @Bean(CALCULATE_AND_LOAD_STATISTICS_JOB)
+    @Bean(FILE_STATS_JOB)
     @Scope("prototype")
-    public Job calculateAndLoadStatisticsJob(JobBuilderFactory jobBuilderFactory) {
-        logger.debug("Building '" + CALCULATE_AND_LOAD_STATISTICS_JOB + "'");
+    public Job fileStatsJob(JobBuilderFactory jobBuilderFactory) {
+        logger.debug("Building '" + FILE_STATS_JOB + "'");
 
         return jobBuilderFactory
-                .get(CALCULATE_AND_LOAD_STATISTICS_JOB)
+                .get(FILE_STATS_JOB)
                 .incrementer(new NewJobIncrementer())
-                .start(calculateAndLoadStatisticsStep)
+                .start(fileStatsStep)
                 .build();
     }
 

src/main/java/uk/ac/ebi/eva/pipeline/configuration/jobs/VariantStatsJobConfiguration.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2024 EMBL - European Bioinformatics Institute
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package uk.ac.ebi.eva.pipeline.configuration.jobs;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
+import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Scope;
+import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.VariantStatsStepConfiguration;
+import uk.ac.ebi.eva.pipeline.parameters.NewJobIncrementer;
+
+import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_JOB;
+import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_STEP;
+
+@Configuration
+@EnableBatchProcessing
+@Import({VariantStatsStepConfiguration.class})
+public class VariantStatsJobConfiguration {
+
+    private static final Logger logger = LoggerFactory.getLogger(VariantStatsJobConfiguration.class);
+
+    @Autowired
+    @Qualifier(VARIANT_STATS_STEP)
+    private Step variantStatsStep;
+
+    @Bean(VARIANT_STATS_JOB)
+    @Scope("prototype")
+    public Job variantStatsJob(JobBuilderFactory jobBuilderFactory) {
+        logger.debug("Building '" + VARIANT_STATS_JOB + "'");
+
+        return jobBuilderFactory
+                .get(VARIANT_STATS_JOB)
+                .incrementer(new NewJobIncrementer())
+                .start(variantStatsStep)
+                .build();
+    }
+
+}

src/main/java/uk/ac/ebi/eva/pipeline/configuration/jobs/steps/FileStatsStepConfiguration.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2024 EMBL - European Bioinformatics Institute
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package uk.ac.ebi.eva.pipeline.configuration.jobs.steps;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
+import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
+import org.springframework.batch.core.configuration.annotation.StepScope;
+import org.springframework.batch.core.step.tasklet.TaskletStep;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import uk.ac.ebi.eva.pipeline.jobs.steps.tasklets.FileStatsTasklet;
+import uk.ac.ebi.eva.pipeline.parameters.ChunkSizeParameters;
+import uk.ac.ebi.eva.pipeline.parameters.DatabaseParameters;
+import uk.ac.ebi.eva.pipeline.parameters.InputParameters;
+import uk.ac.ebi.eva.pipeline.parameters.JobOptions;
+import uk.ac.ebi.eva.utils.TaskletUtils;
+
+import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.FILE_STATS_STEP;
+
+
+@Configuration
+@EnableBatchProcessing
+public class FileStatsStepConfiguration {
+    private static final Logger logger = LoggerFactory.getLogger(FileStatsStepConfiguration.class);
+
+    @Bean
+    @StepScope
+    public FileStatsTasklet fileStatsTasklet(DatabaseParameters databaseParameters,
+                                             MongoTemplate mongoTemplate,
+                                             InputParameters inputParameters,
+                                             ChunkSizeParameters chunkSizeParameters) {
+        return new FileStatsTasklet(databaseParameters, mongoTemplate, inputParameters.getStudyId(),
+                chunkSizeParameters.getChunkSize());
+    }
+
+    @Bean(FILE_STATS_STEP)
+    public TaskletStep fileStatsStep(DatabaseParameters databaseParameters,
+                                     MongoTemplate mongoTemplate,
+                                     InputParameters inputParameters,
+                                     ChunkSizeParameters chunkSizeParameters,
+                                     StepBuilderFactory stepBuilderFactory,
+                                     JobOptions jobOptions) {
+        logger.debug("Building '" + FILE_STATS_STEP + "'");
+
+        return TaskletUtils.generateStep(stepBuilderFactory, FILE_STATS_STEP,
+                fileStatsTasklet(databaseParameters, mongoTemplate, inputParameters, chunkSizeParameters),
+                jobOptions.isAllowStartIfComplete());
+    }
+}

src/main/java/uk/ac/ebi/eva/pipeline/configuration/jobs/steps/CalculateAndLoadStatisticsStepConfiguration.java → src/main/java/uk/ac/ebi/eva/pipeline/configuration/jobs/steps/VariantStatsStepConfiguration.java

@@ -33,26 +33,26 @@
 import uk.ac.ebi.eva.pipeline.configuration.io.writers.VariantStatsWriterConfiguration;
 import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.processors.VariantStatsProcessorConfiguration;
 
-import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.CALCULATE_AND_LOAD_STATISTICS_STEP;
 import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_PROCESSOR;
 import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_READER;
+import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_STEP;
 import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_WRITER;
 
 
 @Configuration
 @EnableBatchProcessing
 @Import({VariantStatsReaderConfiguration.class, VariantStatsWriterConfiguration.class,
         VariantStatsProcessorConfiguration.class, ChunkSizeCompletionPolicyConfiguration.class})
-public class CalculateAndLoadStatisticsStepConfiguration {
+public class VariantStatsStepConfiguration {
 
-    @Bean(CALCULATE_AND_LOAD_STATISTICS_STEP)
-    public Step calculateAndLoadStatisticsStep(
+    @Bean(VARIANT_STATS_STEP)
+    public Step variantStatsStep(
             @Qualifier(VARIANT_STATS_READER) ItemStreamReader<VariantDocument> variantStatsReader,
             @Qualifier(VARIANT_STATS_PROCESSOR) ItemProcessor<VariantDocument, VariantDocument> variantStatsProcessor,
             @Qualifier(VARIANT_STATS_WRITER) ItemWriter<VariantDocument> variantStatsWriter,
             StepBuilderFactory stepBuilderFactory,
             SimpleCompletionPolicy chunkSizeCompletionPolicy) {
-        TaskletStep step = stepBuilderFactory.get(CALCULATE_AND_LOAD_STATISTICS_STEP)
+        TaskletStep step = stepBuilderFactory.get(VARIANT_STATS_STEP)
                 .<VariantDocument, VariantDocument>chunk(chunkSizeCompletionPolicy)
                 .reader(variantStatsReader)
                 .processor(variantStatsProcessor)

src/main/java/uk/ac/ebi/eva/pipeline/io/readers/VariantStatsReader.java

@@ -22,6 +22,8 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static com.mongodb.client.model.Accumulators.sum;
+import static com.mongodb.client.model.Aggregates.group;
 import static com.mongodb.client.model.Aggregates.match;
 import static com.mongodb.client.model.Aggregates.project;
 import static com.mongodb.client.model.Projections.computed;
@@ -92,11 +94,13 @@ private void populateFilesIdAndNumberOfSamplesMap() {
                 computed("fid", "$fid"),
                 computed("numOfSamples", new Document("$size", new Document("$objectToArray", "$samp")))
         ));
+        Bson groupStage = group("$fid", sum("totalNumOfSamples", "$numOfSamples"));
+
         filesIdNumberOfSamplesMap = mongoTemplate.getCollection(databaseParameters.getCollectionFilesName())
-                .aggregate(asList(matchStage, projectStage))
+                .aggregate(asList(matchStage, projectStage, groupStage))
                 .into(new ArrayList<>())
                 .stream()
-                .collect(Collectors.toMap(doc -> doc.getString("fid"), doc -> doc.getInteger("numOfSamples")));
+                .collect(Collectors.toMap(doc -> doc.getString("_id"), doc -> doc.getInteger("totalNumOfSamples")));
     }
 
     public static Map<String, Integer> getFilesIdAndNumberOfSamplesMap() {