feat: parquet export - limit cpu usage

Author: Nolife999

arc-batch/src/main/java/fr/insee/arc/batch/BatchARC.java

@@ -737,6 +737,7 @@ private void initializeBatchLoop() {
 	 * start paralell thread
 	 * 
 	 * @throws ArcException
+	 * @throws InterruptedException 
 	 */
 	private void executeLoopOverPhases() throws ArcException {
 
@@ -762,6 +763,13 @@ private void executeLoopOverPhases() throws ArcException {
 			waitAndClear();
 
 		} while (!exit);
+		
+		// wait for maintenance thread to finish
+		try {
+			maintenance.join();
+		} catch (InterruptedException e) {
+			message("Maintenance thread had been interrupted");
+		}
 
 		message("Fin de la boucle d'itération");
 

arc-core/src/main/java/fr/insee/arc/core/service/p6export/parquet/ParquetDao.java

@@ -8,6 +8,9 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import fr.insee.arc.core.service.p1reception.provider.DirectoriesReception;
+import fr.insee.arc.core.service.p2chargement.bo.NormeRules;
+import fr.insee.arc.core.util.BDParameters;
 import fr.insee.arc.utils.consumer.ThrowingConsumer;
 import fr.insee.arc.utils.dao.DuckdbDao;
 import fr.insee.arc.utils.dao.GenericPreparedStatementBuilder;
@@ -23,6 +26,10 @@ public class ParquetDao {
 
 	private static final Logger LOGGER = LogManager.getLogger(ParquetDao.class);
 
+	// Ratio MAX_PARALLEL_WORKERS parameters
+	// used not to over throttle input database or s3
+	private static final float RATIO_OF_CPU_USED_TO_EXPORT_PARQUET = 0.5F;
+	
 	// parquet file format as "file.parquet"
 	private static final String PARQUET_FILE_EXTENSION = ".parquet";
 
@@ -69,6 +76,8 @@ public void exportToParquet(List<TableToRetrieve> tables, String outputDirectory
 		{
 			addParquetEncryptionKeyInDuckDb(connection);
 
+			defineNumberOfThreadThread(connection);
+			
 			// create output directory
 			FileUtilsArc.createDirIfNotexist(outputDirectory);
 
@@ -79,10 +88,26 @@ public void exportToParquet(List<TableToRetrieve> tables, String outputDirectory
 			}
 		};
 
+		
 		duckdbDao.executeOnDuckdb(exportToParquetOperation);
 
 	}
 
+	private void defineNumberOfThreadThread(Connection connection) throws ArcException {
+		
+		BDParameters bdParameters = new BDParameters(ArcDatabase.COORDINATOR);
+		
+		int maxParallelWorkers = (int) (bdParameters.getInt(null,"ApiService.MAX_PARALLEL_WORKERS", 4) * RATIO_OF_CPU_USED_TO_EXPORT_PARQUET) ;
+		maxParallelWorkers = (maxParallelWorkers < 1)? 1 : maxParallelWorkers;
+		
+		GenericPreparedStatementBuilder query = new GenericPreparedStatementBuilder();
+		query.append("SET threads=").append(query.quoteInt(maxParallelWorkers)).append(";");
+
+		duckdbDao.executeQuery(connection, query);		
+
+	}
+	
+	
 	private void addParquetEncryptionKeyInDuckDb(Connection connection) throws ArcException {
 
 		if (encryptionKey == null) {

arc-core/src/test/java/fr/insee/arc/core/service/p6export/parquet/ParquetDaoTest.java

@@ -99,6 +99,15 @@ private void createTestTable(Connection connection, String testTableName) throws
 
 		// test query on
 		GenericPreparedStatementBuilder query = new GenericPreparedStatementBuilder();
+		
+		query.build(SQL.CREATE, SQL.SCHEMA, SQL.IF_NOT_EXISTS, "arc", SQL.END_QUERY); 
+		
+		query.build(SQL.CREATE, SQL.TABLE, "arc.parameter", SQL.AS, SQL.SELECT);
+		query.append("'ApiService.MAX_PARALLEL_WORKERS'::text as key");
+		query.append(",'12'::text as val");
+		query.append(SQL.END_QUERY);
+
+		
 		query.build(SQL.CREATE, SQL.TABLE, testTableName, SQL.AS, SQL.SELECT);
 		query.append("'string'::text as column_string");
 		query.append(",12::int as column_int4");
@@ -110,7 +119,8 @@ private void createTestTable(Connection connection, String testTableName) throws
 		query.append(",array[8, 9, 10]::int[] as column_array_int8");
 		query.append(",array[current_timestamp, current_timestamp] as column_array_timestamp");
 		query.append(",array[current_date, current_date] as column_array_date");
-
+		query.append(SQL.END_QUERY);
+		
 		UtilitaireDao.get(0).executeRequest(connection, query);
 	}
 

arc-ws/src/main/java/fr/insee/arc/ws/services/importServlet/dao/ClientDao.java

@@ -14,7 +14,6 @@
 import fr.insee.arc.core.dataobjects.ColumnEnum;
 import fr.insee.arc.core.dataobjects.ViewEnum;
 import fr.insee.arc.core.model.ExportOption;
-import fr.insee.arc.core.model.TraitementEtape;
 import fr.insee.arc.core.model.TraitementEtat;
 import fr.insee.arc.core.model.TraitementPhase;
 import fr.insee.arc.core.service.global.dao.TableNaming;