Merge pull request #461 from dadosfera/feat/qualify

FEAT: using batch route to update pipeline

src/modules/inputs/inputs.controller.ts

@@ -99,6 +99,7 @@ export class InputsController {
       customer: info.customer,
     });
 
+    this.logger.info(JSON.stringify(body))
     const response = await this.inputService.create({ body, info });
 
     return response;

src/modules/pipelinesV2/pipelines.service.ts

@@ -356,125 +356,92 @@ export class PipelinesService implements OnModuleInit {
       info
     )
 
-    const requests = [];
-
     this.logger.info('Dynamo Response', updateInputResponse);
+    const jobsUpdated = [];
 
     for (const [index, table] of updateInputDTO.tables.entries()) {
-      const id = `${pipelineIdFormat}_${index}`;
+      const jobUpdate = {
+        job_id: `${pipelineIdFormat}_${index}`,
+      }
+
+      if (table.memory) {
+        jobUpdate["memory"] = {
+          amount: table.memory * 1000
+        }
+      }
+      
       this.logger.info('Updating input reference for table', table.name);
-      const body = {}
+      let hasUpdateSyncMode = false;
+
+      const jobSyncMode = {}
+
       if (table.columns) {
-        body['column_include_list'] = table.columns;
+        hasUpdateSyncMode = true;
+        jobSyncMode['column_include_list'] = table.columns;
       }
 
       if (table.reference_column) {
-        body['incremental_column_name'] = table.reference_column.name;
-        body['incremental_column_type'] = table.reference_column.type;
+        hasUpdateSyncMode = true;
+        jobSyncMode['incremental_column_name'] = table.reference_column.name;
+        jobSyncMode['incremental_column_type'] = table.reference_column.type;
       }
 
       if (table.identifier_columns) {
-        body['primary_keys'] = table.identifier_columns;
-      }
-
-      this.logger.info('Request body', body);
-      const updateCollumns = this.platformAPI.proxy(
-        'PATCH',
-        `/jobs/${id}/input`,
-        user,
-        body
-      )
-      requests.push(updateCollumns);
-
-      if (table.memory) {
-        this.logger.info('Updating memory allocation for table', table.name);
-        const updateMemory = this.platformAPI.proxy(
-          'PUT',
-          `/jobs/${id}/memory`,
-          user,
-          {
-            amount: table.memory
-          }
-        )
-        requests.push(updateMemory);
+        hasUpdateSyncMode = true;
+        jobSyncMode['primary_keys'] = table.identifier_columns;
       }
 
       if (table.type) {
-          const updateSyncMode = this.updatePipelineSyncMode(table, id, user);
-          requests.push(updateSyncMode);
+        hasUpdateSyncMode = true;
+        jobSyncMode['target_load_type'] = table.type;
+
       }
-    }
 
-    this.logger.info('Create Platform Request for each JOB');
+      if(hasUpdateSyncMode) {
+        jobUpdate["sync_mode"] = jobSyncMode;
+      }
 
-    if (updateInputDTO.cron) {
-      const crnUpdatedRequest = new Promise(async (resolve, reject) => {
-        const response = await this.updatePipelineCron(updateInputDTO.cron, pipelineIdFormat, user);
-
-        if (response.error) {
-          this.logger.error('Error updating pipeline cron', response.error);
-          return reject(new ErrorBuilder(response.error));
+      if (Object.keys(table.destinations).length > 1) {
+        let hasChanges = false
+        const jobRenameTables = {
+            raw: {},
+            qualify: {}
         }
-        this.logger.error('Pipeline cron updated successfully', response);
-        return resolve(response);
-      });
-      requests.push(crnUpdatedRequest);
+
+        if (Object.keys(table.destinations.raw).length > 1) {
+          hasChanges = true;
+          jobRenameTables.raw = table.destinations.raw;
+        }
+
+        if (Object.keys(table.destinations.qualify).length > 1) {
+          hasChanges = true;
+          jobRenameTables.qualify = table.destinations.qualify;
+        }
+
+        if (hasChanges) {
+          jobUpdate['rename_tables'] = jobRenameTables;
+        }
+      }
+
+      jobsUpdated.push(jobUpdate);
     }
 
-    this.logger.info('Executing all request for the platform api');
+    this.logger.info('Request body:' + JSON.stringify({
+      jobs_updated: jobsUpdated
+    }));
 
-    const results = await Promise.allSettled(requests);
-    this.logger.info('Platform api response', results);
+    const response = await this.platformAPI.proxy(
+      'POST',
+        `/pipelines/${pipelineIdFormat}/batch-update`,
+      user,
+      {
+        job_updates: jobsUpdated
+      }
+    )
+    this.logger.info('Platform api response: ' + JSON.stringify(response));
 
     return updateInputResponse;
     
   }
 
-  private async updatePipelineSyncMode(table: UpdateTableDTO,  pipelineId: string, user: RequestUser) {
-    const body = {
-      target_load_type: table.type
-    }
-
-    if (table.type === 'incremental_with_qualify') {
-      body['incremental_column_name'] = table.reference_column.name;
-      body['incremental_column_type'] = table.reference_column.type;
-      body['primary_keys'] = table.identifier_columns;
-    }
-
-    if (table.type === 'incremental') {
-      body['incremental_column_name'] = table.reference_column.name;
-      body['incremental_column_type'] = table.reference_column.type;
-    }
-
-    this.logger.info('Updating pipeline sync mode', {
-      pipelineId,
-      body
-    });
-
-    return this.platformAPI.proxy(
-      "POST",
-      `/jobs/jdbc/${pipelineId}/sync-mode`,
-      user,
-      body
-    )
-  }
-
-  private async updatePipelineCron(cron: string, pipelineId: string, user: RequestUser) {
-    try {
-      const response = await this.platformAPI.proxy(
-        'PATCH',
-        `/pipeline/${pipelineId}`,
-        user,
-        {
-          cron
-        }
-      );
-
-      return response
-    } catch (error) {
-        return {
-          error: error.message
-        }
-    }
-  }
 }

src/modules/platform-api/platform-api.controller.ts

@@ -1066,8 +1066,8 @@ export class PlatformApiController {
       const rollbackSteps: Array<() => Promise<void>> = [];
 
       try {
-        const dataAsset = await this.elasticsearchService.findDataAssetByPipelineAndTable(
-          user.customer_name, pipelineId, oldTableName, oldTableSchema,
+        const dataAsset = await this.elasticsearchService.findDataAssetByTable(
+          user.customer_name, oldTableName, oldTableSchema,
         );
 
         if (!dataAsset) {
@@ -1086,26 +1086,31 @@ export class PlatformApiController {
           { name: oldTableName, table_name: oldTableName, table_schema: oldTableSchema, display_name: oldTableName },
         ));
 
+        const newTableNameUpper = newValues.table_name.toUpperCase();
+        const newTableSchemaUpper = newValues.table_schema.toUpperCase();
+        const oldTableNameUpper = oldTableName.toUpperCase();
+        const oldTableSchemaUpper = oldTableSchema.toUpperCase();
+
         // Nimbus table-metadata
         if (nimbusId) {
-          await this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, newValues);
-          rollbackSteps.push(() => this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, oldValues));
+          await this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, { table_name: newTableNameUpper, table_schema: newTableSchemaUpper });
+          rollbackSteps.push(() => this.catalogService.renameTableOnNimbus(nimbusUrl, nimbusId, { table_name: oldTableNameUpper, table_schema: oldTableSchemaUpper }));
         }
 
         // Nimbus column-metadata
         await this.catalogService.renameColumnMetadataOnNimbus(
-          nimbusUrl, databaseName, oldTableName, oldTableSchema, newValues.table_name, newValues.table_schema,
+          nimbusUrl, databaseName, oldTableNameUpper, oldTableSchemaUpper, newTableNameUpper, newTableSchemaUpper,
         );
         rollbackSteps.push(() => this.catalogService.renameColumnMetadataOnNimbus(
-          nimbusUrl, databaseName, newValues.table_name, newValues.table_schema, oldTableName, oldTableSchema,
+          nimbusUrl, databaseName, newTableNameUpper, newTableSchemaUpper, oldTableNameUpper, oldTableSchemaUpper,
         ));
 
         // Nimbus data-preview
         await this.catalogService.renameDataPreviewOnNimbus(
-          nimbusUrl, databaseName, oldTableName, oldTableSchema, newValues.table_name, newValues.table_schema,
+          nimbusUrl, databaseName, oldTableNameUpper, oldTableSchemaUpper, newTableNameUpper, newTableSchemaUpper,
         );
         rollbackSteps.push(() => this.catalogService.renameDataPreviewOnNimbus(
-          nimbusUrl, databaseName, newValues.table_name, newValues.table_schema, oldTableName, oldTableSchema,
+          nimbusUrl, databaseName, newTableNameUpper, newTableSchemaUpper, oldTableNameUpper, oldTableSchemaUpper,
         ));
 
         this.logger.info(`Synced catalog rename for ${key}`, { jobId, oldTableName, newTableName: newValues.table_name });

src/services/elasticsearch/elasticsearch.service.ts

@@ -362,9 +362,8 @@ export class ElasticsearchService {
     return `${customerName}_data_assets_catalog`;
   }
 
-  async findDataAssetByPipelineAndTable(
+  async findDataAssetByTable(
     customerName: string,
-    pipelineId: string,
     tableName: string,
     tableSchema: string,
   ): Promise<{ id: string; nimbus_id: number | null; [key: string]: any } | null> {
@@ -372,7 +371,6 @@ export class ElasticsearchService {
 
     this.logger.info('Elasticsearch: Searching data asset', {
       index,
-      pipelineId,
       tableName,
       tableSchema,
     });
@@ -382,9 +380,8 @@ export class ElasticsearchService {
         query: {
           bool: {
             must: [
-              { term: { 'pipeline_id.keyword': pipelineId } },
-              { term: { 'table_name.keyword': tableName } },
-              { term: { 'table_schema.keyword': tableSchema } },
+              { term: { 'table_name.keyword': tableName.toUpperCase() } },
+              { term: { 'table_schema.keyword': tableSchema.toUpperCase() } },
             ],
           },
         },
@@ -393,13 +390,13 @@ export class ElasticsearchService {
 
       const hits = response.data.hits?.hits || [];
       if (hits.length === 0) {
-        this.logger.warn('Elasticsearch: Data asset not found', { pipelineId, tableName, tableSchema, index });
+        this.logger.warn('Elasticsearch: Data asset not found', { tableName, tableSchema, index });
         return null;
       }
 
       return { ...hits[0]._source, _es_id: hits[0]._id };
     } catch (error) {
-      this.handleError('findDataAssetByPipelineAndTable', error, { pipelineId, tableName, tableSchema, index });
+      this.handleError('findDataAssetByTable', error, { tableName, tableSchema, index });
       throw error;
     }
   }