chore: use refresh start time to refresh base view (#1457)

Author: dengmingtong

src/analytics/lambdas/refresh-materialized-views-workflow/refresh-basic-view.ts

@@ -26,6 +26,9 @@ export interface RefreshBasicViewEvent {
     appId: string;
     timezone: string;
   };
+  originalInput: {
+    refreshStartTime: string;
+  };
 }
 
 /**
@@ -51,14 +54,21 @@ export const handler = async (event: RefreshBasicViewEvent) => {
 
   const sqlStatements: string[] = [];
   const timezoneWithAppId = event.timezoneWithAppId;
+  const originalInput = event.originalInput;
   const viewName = event.view.name;
 
-  const dataFreshnessInHour = process.env.DATA_REFRESHNESS_IN_HOUR!;
-
   try {
     let queryId : string | undefined;
     const type = event.view.type;
     if (type === 'custom-mv') {
+      let dataFreshnessInHour = process.env.DATA_REFRESHNESS_IN_HOUR!;
+      if (originalInput.refreshStartTime) {
+        let dataFreshnessInHourNumber = calculateHoursAgo(originalInput.refreshStartTime);
+        if (dataFreshnessInHourNumber > 0) {
+          dataFreshnessInHour = dataFreshnessInHourNumber.toString();
+        }
+      }
+      logger.info('dataFreshnessInHour', { dataFreshnessInHour });
       sqlStatements.push(`CALL ${timezoneWithAppId.appId}.${viewName}(NULL, NULL, ${dataFreshnessInHour});`);
     } else {
       sqlStatements.push(`REFRESH MATERIALIZED VIEW ${timezoneWithAppId.appId}.${viewName};`);
@@ -79,4 +89,11 @@ export const handler = async (event: RefreshBasicViewEvent) => {
     logger.error('Error when refresh mv:', { err });
     throw err;
   }
-};
+};
+
+function calculateHoursAgo(refreshStartTime: string) {
+  const currentDate = new Date();
+  const diffInMilliseconds = currentDate.getTime() - parseInt(refreshStartTime);
+  const diffInHours = Math.floor(diffInMilliseconds / (1000 * 60 * 60) ) + 1;
+  return diffInHours;
+}

src/analytics/private/refresh-materialized-views-workflow.ts

@@ -98,6 +98,7 @@ export class RefreshMaterializedViewsWorkflow extends Construct {
       payload: TaskInput.fromObject({
         'view.$': '$.view',
         'timezoneWithAppId.$': '$.timezoneWithAppId',
+        'originalInput.$': '$$.Execution.Input',
       }),
       outputPath: '$.Payload',
     });

test/analytics/analytics-on-redshift/lambda/refresh-materialized-views-workflow/refresh-basic-view.test.ts

@@ -26,7 +26,10 @@ const refreshBasicViewEvent: RefreshBasicViewEvent = {
   },
   timezoneWithAppId: {
     appId: 'app1',
-    timezone: 'America/Noronha',
+    timezone: 'UTC',
+  },
+  originalInput: {
+    refreshStartTime: '',
   },
 };
 
@@ -62,8 +65,86 @@ describe('Lambda - do refresh job in Redshift Serverless', () => {
     });
   });
 
+  test('Executed Redshift refresh custom-mv command', async () => {
+    const exeuteId = 'Id-1';
+    refreshBasicViewEvent.view.type = 'custom-mv';
+    refreshBasicViewEvent.originalInput = {
+      refreshStartTime: '1715470905000',
+    };
+    jest
+      .useFakeTimers()
+      .setSystemTime(1715490905000);
+    const dataFreshnessInHour = 6;
+    redshiftDataMock.on(ExecuteStatementCommand).resolvesOnce({ Id: exeuteId });
+    const resp = await handler(refreshBasicViewEvent);
+    expect(resp).toEqual({
+      detail: {
+        viewName: refreshBasicViewEvent.view.name,
+        queryId: exeuteId,
+      },
+      timezoneWithAppId: refreshBasicViewEvent.timezoneWithAppId,
+    });
+    expect(redshiftDataMock).toHaveReceivedCommandWith(ExecuteStatementCommand, {
+      WorkgroupName: workGroupName,
+      Sql: `CALL ${refreshBasicViewEvent.timezoneWithAppId.appId}.${refreshBasicViewEvent.view.name}(NULL, NULL, ${dataFreshnessInHour});`,
+      Database: expect.any(String),
+    });
+    jest.useRealTimers();
+  });
+
+  test('Executed Redshift refresh custom-mv command with RefreshStartTime bigger than current', async () => {
+    const exeuteId = 'Id-1';
+    refreshBasicViewEvent.view.type = 'custom-mv';
+    refreshBasicViewEvent.originalInput = {
+      refreshStartTime: '1715470905000',
+    };
+    jest
+      .useFakeTimers()
+      .setSystemTime(1715450905000);
+    const dataFreshnessInHour = 72;
+    redshiftDataMock.on(ExecuteStatementCommand).resolvesOnce({ Id: exeuteId });
+    const resp = await handler(refreshBasicViewEvent);
+    expect(resp).toEqual({
+      detail: {
+        viewName: refreshBasicViewEvent.view.name,
+        queryId: exeuteId,
+      },
+      timezoneWithAppId: refreshBasicViewEvent.timezoneWithAppId,
+    });
+    expect(redshiftDataMock).toHaveReceivedCommandWith(ExecuteStatementCommand, {
+      WorkgroupName: workGroupName,
+      Sql: `CALL ${refreshBasicViewEvent.timezoneWithAppId.appId}.${refreshBasicViewEvent.view.name}(NULL, NULL, ${dataFreshnessInHour});`,
+      Database: expect.any(String),
+    });
+    jest.useRealTimers();
+  });
+
+  test('Executed Redshift refresh custom-mv command without input refreshStartTime', async () => {
+    const exeuteId = 'Id-1';
+    refreshBasicViewEvent.view.type = 'custom-mv';
+    refreshBasicViewEvent.originalInput = {
+      refreshStartTime: '',
+    };
+    const dataFreshnessInHour = 72;
+    redshiftDataMock.on(ExecuteStatementCommand).resolvesOnce({ Id: exeuteId });
+    const resp = await handler(refreshBasicViewEvent);
+    expect(resp).toEqual({
+      detail: {
+        viewName: refreshBasicViewEvent.view.name,
+        queryId: exeuteId,
+      },
+      timezoneWithAppId: refreshBasicViewEvent.timezoneWithAppId,
+    });
+    expect(redshiftDataMock).toHaveReceivedCommandWith(ExecuteStatementCommand, {
+      WorkgroupName: workGroupName,
+      Sql: `CALL ${refreshBasicViewEvent.timezoneWithAppId.appId}.${refreshBasicViewEvent.view.name}(NULL, NULL, ${dataFreshnessInHour});`,
+      Database: expect.any(String),
+    });
+  });
+
   test('Execute command error in Redshift when doing refresh', async () => {
     redshiftDataMock.on(ExecuteStatementCommand).rejects();
+    refreshBasicViewEvent.view.type = 'basic';
     try {
       await handler(refreshBasicViewEvent);
       fail('The error in executing statement of Redshift data was caught');
@@ -79,7 +160,6 @@ describe('Lambda - do refresh job in Redshift Serverless', () => {
     redshiftDataMock.on(ExecuteStatementCommand)
       .rejectsOnce()
       .resolves({ Id: exeuteId });
-
     const resp = await handler(refreshBasicViewEvent);
     expect(resp).toEqual({
       detail: {