feat: Allow persisting multiple pre-aggregation structure versions to support staging pre-aggregation warm-up environments and multiple timezones

Author: paveltiunov

packages/cubejs-query-orchestrator/driver/BaseDriver.js

@@ -4,7 +4,7 @@ const { cancelCombinator } = require('./utils');
 const sortByKeys = (unordered) => {
   const ordered = {};
 
-  Object.keys(unordered).sort().forEach(function(key) {
+  Object.keys(unordered).sort().forEach((key) => {
     ordered[key] = unordered[key];
   });
 
@@ -13,16 +13,16 @@ const sortByKeys = (unordered) => {
 
 const DbTypeToGenericType = {
   'timestamp without time zone': 'timestamp',
-  'integer': 'int',
+  integer: 'int',
   'character varying': 'text',
-  'varchar': 'text',
-  'text': 'text',
-  'string': 'text',
-  'boolean': 'boolean',
-  'bigint': 'bigint',
-  'time': 'string',
-  'datetime': 'timestamp',
-  'date': 'date',
+  varchar: 'text',
+  text: 'text',
+  string: 'text',
+  boolean: 'boolean',
+  bigint: 'bigint',
+  time: 'string',
+  datetime: 'timestamp',
+  date: 'date',
   'double precision': 'decimal'
 };
 
@@ -59,7 +59,7 @@ class BaseDriver {
 
     const reduceCb = (result, i) => {
       let schema = (result[i.table_schema] || {});
-      let tables = (schema[i.table_name] || []);
+      const tables = (schema[i.table_name] || []);
 
       tables.push({ name: i.column_name, type: i.data_type, attributes: i.key_type ? ['primaryKey'] : [] });
 
@@ -79,10 +79,11 @@ class BaseDriver {
       `SELECT schema_name FROM information_schema.schemata WHERE schema_name = ${this.param(0)}`,
       [schemaName]
     ).then((schemas) => {
-        if (schemas.length === 0) {
-          return this.query("CREATE SCHEMA IF NOT EXISTS " + schemaName);
-        }
-      });
+      if (schemas.length === 0) {
+        return this.query(`CREATE SCHEMA IF NOT EXISTS ${schemaName}`);
+      }
+      return null;
+    });
   }
 
   getTablesQuery(schemaName) {
@@ -132,6 +133,7 @@ class BaseDriver {
     }
   }
 
+  // eslint-disable-next-line no-unused-vars
   toColumnValue(value, genericType) {
     return value;
   }

packages/cubejs-query-orchestrator/orchestrator/PreAggregations.js

@@ -17,33 +17,35 @@ function version(cacheKey) {
   for (let i = 0; i < 5; i++) {
     const byte = digestBuffer.readUInt8(i);
     shiftCounter += 8;
+    // eslint-disable-next-line operator-assignment,no-bitwise
     residue = (byte << (shiftCounter - 8)) | residue;
+    // eslint-disable-next-line no-bitwise
     while (residue >> 5) {
       result += hashCharset.charAt(residue % 32);
       shiftCounter -= 5;
+      // eslint-disable-next-line operator-assignment,no-bitwise
       residue = residue >> 5;
     }
   }
   result += hashCharset.charAt(residue % 32);
   return result;
 }
 
-const tablesToVersionEntries = (schema, tables) => {
-  return R.sortBy(
-    table => -table.last_updated_at,
-    tables.map(table => {
-      const match = (table.table_name || table.TABLE_NAME).match(/(.+)_(.+)_(.+)_(.+)/);
-      if (match) {
-        return {
-          table_name: `${schema}.${match[1]}`,
-          content_version: match[2],
-          structure_version: match[3],
-          last_updated_at: parseInt(match[4], 10)
-        }
-      }
-    }).filter(R.identity)
-  )
-};
+const tablesToVersionEntries = (schema, tables) => R.sortBy(
+  table => -table.last_updated_at,
+  tables.map(table => {
+    const match = (table.table_name || table.TABLE_NAME).match(/(.+)_(.+)_(.+)_(.+)/);
+    if (match) {
+      return {
+        table_name: `${schema}.${match[1]}`,
+        content_version: match[2],
+        structure_version: match[3],
+        last_updated_at: parseInt(match[4], 10)
+      };
+    }
+    return null;
+  }).filter(R.identity)
+);
 
 class PreAggregationLoadCache {
   constructor(redisPrefix, clientFactory, queryCache, preAggregations, options) {
@@ -177,6 +179,7 @@ class PreAggregationLoader {
     this.waitForRenew = options.waitForRenew;
     this.externalDriverFactory = preAggregations.externalDriverFactory;
     this.requestId = options.requestId;
+    this.structureVersionPersistTime = preAggregations.structureVersionPersistTime;
   }
 
   async loadPreAggregation() {
@@ -225,7 +228,7 @@ class PreAggregationLoader {
 
     const versionEntries = await this.loadCache.getVersionEntries(this.preAggregation);
 
-    const getVersionEntryByContentVersion = (versionEntries) => versionEntries.find(
+    const getVersionEntryByContentVersion = (entries) => entries.find(
       v => v.table_name === this.preAggregation.tableName && v.content_version === contentVersion
     );
 
@@ -237,6 +240,7 @@ class PreAggregationLoader {
     // TODO this check can be redundant due to structure version is already checked in loadPreAggregation()
     if (
       !this.waitForRenew &&
+      // eslint-disable-next-line no-use-before-define
       await this.loadCache.getQueryStage(PreAggregations.preAggregationQueryCacheKey(this.preAggregation))
     ) {
       const versionEntryByStructureVersion = versionEntries.find(
@@ -253,7 +257,8 @@ class PreAggregationLoader {
         await this.driverFactory();
       await client.createSchemaIfNotExists(this.preAggregation.preAggregationsSchema);
     }
-    const versionEntry = versionEntries.find(e => e.table_name === this.preAggregation.tableName); // TODO can be array instead of last
+    // TODO can be array instead of last
+    const versionEntry = versionEntries.find(e => e.table_name === this.preAggregation.tableName);
     const newVersionEntry = {
       table_name: this.preAggregation.tableName,
       structure_version: structureVersion,
@@ -363,6 +368,7 @@ class PreAggregationLoader {
         requestId: this.requestId
       },
       priority,
+      // eslint-disable-next-line no-use-before-define
       { stageQueryKey: PreAggregations.preAggregationQueryCacheKey(this.preAggregation), requestId: this.requestId }
     );
   }
@@ -526,8 +532,17 @@ class PreAggregationLoader {
       R.toPairs,
       R.map(p => p[1][0])
     )(versionEntries);
+
+    const structureVersionsToSave = R.pipe(
+      R.filter(v => new Date().getTime() - v.last_updated_at < this.structureVersionPersistTime * 1000),
+      R.groupBy(v => `${v.table_name}_${v.structure_version}`),
+      R.toPairs,
+      R.map(p => p[1][0])
+    )(versionEntries);
+
     const tablesToSave =
       (await this.preAggregations.tablesUsed())
+        .concat(structureVersionsToSave.map(v => this.targetTableName(v)))
         .concat(versionEntriesToSave.map(v => this.targetTableName(v)))
         .concat([justCreatedTable]);
     const toDrop = actualTables
@@ -552,14 +567,16 @@ class PreAggregations {
       new RedisCacheDriver({ pool: options.redisPool }) :
       new LocalCacheDriver();
     this.externalDriverFactory = options.externalDriverFactory;
+    this.structureVersionPersistTime = options.structureVersionPersistTime || 60 * 60 * 24 * 30;
+    this.usedTablePersistTime = options.usedTablePersistTime || 600;
   }
 
   tablesUsedRedisKey(tableName) {
     return `SQL_PRE_AGGREGATIONS_${this.redisPrefix}_TABLES_USED_${tableName}`;
   }
 
   async addTableUsed(tableName) {
-    return this.cacheDriver.set(this.tablesUsedRedisKey(tableName), true, 600);
+    return this.cacheDriver.set(this.tablesUsedRedisKey(tableName), true, this.usedTablePersistTime);
   }
 
   async tablesUsed() {

packages/cubejs-query-orchestrator/package.json

@@ -12,7 +12,8 @@
     "node": ">=8.11.1"
   },
   "scripts": {
-    "test": "jest --runInBand --verbose"
+    "test": "jest --runInBand --verbose",
+    "lint": "eslint orchestrator/*.js driver/*.js"
   },
   "dependencies": {
     "generic-pool": "^3.7.1",
@@ -21,6 +22,8 @@
   },
   "devDependencies": {
     "eslint": "^6.8.0",
+    "eslint-config-airbnb-base": "^13.1.0",
+    "eslint-plugin-import": "^2.16.0",
     "eslint-plugin-node": "^5.2.1",
     "jest": "^25.1.0"
   },

packages/cubejs-query-orchestrator/test/QueryOrchestrator.test.js

@@ -35,7 +35,7 @@ class MockDriver {
   }
 
   async dropTable(tableName) {
-    this.tables = this.tables.filter(t => t !== tableName.split('.')[1]);
+    this.tables = this.tables.filter(t => t !== tableName);
     return this.query(`DROP TABLE ${tableName}`);
   }
 }
@@ -49,7 +49,8 @@ describe('QueryOrchestrator', () => {
       preAggregationsOptions: {
         queueOptions: {
           executionTimeout: 1
-        }
+        },
+        usedTablePersistTime: 1
       }
     }
   );
@@ -162,4 +163,64 @@ describe('QueryOrchestrator', () => {
     }
     expect(mockDriver.cancelledQueries[0]).toMatch(/orders_too_big/);
   });
+
+  test('save structure versions', async () => {
+    mockDriver.tables = [];
+    await queryOrchestrator.fetchQuery({
+      query: `SELECT * FROM stb_pre_aggregations.orders`,
+      values: [],
+      cacheKeyQueries: {
+        renewalThreshold: 21600,
+        queries: []
+      },
+      preAggregations: [{
+        preAggregationsSchema: "stb_pre_aggregations",
+        tableName: "stb_pre_aggregations.orders",
+        loadSql: ["CREATE TABLE stb_pre_aggregations.orders AS SELECT * FROM public.orders", []],
+        invalidateKeyQueries: [["SELECT 1", []]]
+      }],
+      renewQuery: true,
+      requestId: 'save structure versions'
+    });
+
+    await queryOrchestrator.fetchQuery({
+      query: `SELECT * FROM stb_pre_aggregations.orders`,
+      values: [],
+      cacheKeyQueries: {
+        renewalThreshold: 21600,
+        queries: []
+      },
+      preAggregations: [{
+        preAggregationsSchema: "stb_pre_aggregations",
+        tableName: "stb_pre_aggregations.orders",
+        loadSql: ["CREATE TABLE stb_pre_aggregations.orders AS SELECT * FROM public.orders1", []],
+        invalidateKeyQueries: [["SELECT 1", []]]
+      }],
+      renewQuery: true,
+      requestId: 'save structure versions'
+    });
+
+    await new Promise(resolve => setTimeout(() => resolve(), 1000));
+
+    for (let i = 0; i < 5; i++) {
+      await queryOrchestrator.fetchQuery({
+        query: `SELECT * FROM stb_pre_aggregations.orders`,
+        values: [],
+        cacheKeyQueries: {
+          renewalThreshold: 21600,
+          queries: []
+        },
+        preAggregations: [{
+          preAggregationsSchema: "stb_pre_aggregations",
+          tableName: "stb_pre_aggregations.orders",
+          loadSql: ["CREATE TABLE stb_pre_aggregations.orders AS SELECT * FROM public.orders", []],
+          invalidateKeyQueries: [["SELECT 2", []]]
+        }],
+        renewQuery: true,
+        requestId: 'save structure versions'
+      });
+    }
+    expect(mockDriver.tables).toContainEqual(expect.stringMatching(/orders_f5v4jw3p_4eysppzt/));
+    expect(mockDriver.tables).toContainEqual(expect.stringMatching(/orders_mjooke4_ezlvkhjl/));
+  });
 });