Merge pull request #283 from Canner/feature/caching-layer-persistent

onlyjackfrost · web-flow · commit fc74f5596f7e · 2023-08-18T18:02:26.000+08:00
Fix: caching layer persistent
diff --git a/packages/core/src/containers/modules/dataSource.ts b/packages/core/src/containers/modules/dataSource.ts
@@ -8,7 +8,11 @@ import {
 import { Profile } from '../../models/profile';
 import { ClassType } from '../../lib/utils/module';
 import { ConfigurationError } from '@vulcan-sql/core/utils';
-import { ICacheLayerOptions, cacheProfileName } from '@vulcan-sql/core/models';
+import {
+  ICacheLayerOptions,
+  cacheLayerPersistentFileName,
+  cacheProfileName,
+} from '@vulcan-sql/core/models';
 import 'reflect-metadata';
 
 export const dataSourceModule = (
@@ -26,6 +30,7 @@ export const dataSourceModule = (
         type: options.loader!.toLocaleLowerCase(),
         // allow '*' to make every user request could use the cache-layer data source.
         allow: '*',
+        connection: { ['persistent-path']: cacheLayerPersistentFileName },
       } as Profile);
     }
 
diff --git a/packages/core/src/lib/cache-layer/cacheLayerLoader.ts b/packages/core/src/lib/cache-layer/cacheLayerLoader.ts
@@ -43,31 +43,40 @@ export class CacheLayerLoader implements ICacheLayerLoader {
     templateName: string,
     cache: CacheLayerInfo
   ): Promise<void> {
-    const { cacheTableName, sql, profile, indexes } = cache;
+    const { cacheTableName, sql, profile, indexes, folderSubpath } = cache;
     const type = this.options.type!;
     const dataSource = this.dataSourceFactory(profile);
 
     // generate directory for cache file path to export
     // format => [folderPath]/[schema.templateSource]/[profileName]/[cacheTableName]]/[timestamp]
+    const subpath = folderSubpath || moment.utc().format('YYYYMMDDHHmmss');
     const directory = path.resolve(
       this.options.folderPath!,
       templateName,
       profile,
       cacheTableName,
-      moment.utc().format('YYYYMMDDHHmmss')
+      subpath
     );
-
-    if (!fs.existsSync(directory!))
-      fs.mkdirSync(directory!, { recursive: true });
-
-    // 1. export to cache files according to each schema set the cache value
-    this.logger.debug(`Start to export to ${type} file in "${directory}"`);
-    await dataSource.export({
-      sql,
-      directory,
-      profileName: profile,
-      type,
-    });
+    const parquetFiles = this.getParquetFiles(directory);
+    if (!parquetFiles.length) {
+      if (!fs.existsSync(directory!)) {
+        fs.mkdirSync(directory!, { recursive: true });
+      }
+      // 1. export to cache files according to each schema set the cache value
+      this.logger.debug(`Start to export to ${type} file in "${directory}"`);
+      await dataSource.export({
+        sql,
+        directory,
+        profileName: profile,
+        type,
+      });
+    } else {
+      this.logger.debug(
+        `Parquet file \n ${parquetFiles.join(
+          '\n  '
+        )} found in ${directory}, skip export`
+      );
+    }
     this.logger.debug(`Start to load ${cacheTableName} in "${directory}"`);
     // 2. load the files to cache data source
     await this.cacheStorage.import({
@@ -81,4 +90,11 @@ export class CacheLayerLoader implements ICacheLayerLoader {
       indexes,
     });
   }
+
+  private getParquetFiles(directory: string): string[] {
+    if (!directory || !fs.existsSync(directory)) return [];
+    const files = fs.readdirSync(directory);
+    const parquetFiles = files.filter((file) => /\.parquet$/.test(file));
+    return parquetFiles;
+  }
 }
diff --git a/packages/core/src/models/artifact.ts b/packages/core/src/models/artifact.ts
@@ -114,6 +114,8 @@ export class CacheLayerInfo {
   refreshExpression?: RefreshExpression;
   // index key name -> index column
   indexes?: Record<string, string>;
+  // cache folder subpath
+  folderSubpath?: string;
 }
 
 export class APISchema {
diff --git a/packages/core/src/models/cacheLayerOptions.ts b/packages/core/src/models/cacheLayerOptions.ts
@@ -19,6 +19,11 @@ export interface ICacheLayerOptions {
 // The cache layer profile name is used to load the cache data to table name from cache files
 export const cacheProfileName = 'vulcan.cache';
 
+// The cache layer persistent file name, if the file name is set to ":memory:", it will use in-memory database
+export const cacheLayerPersistentFileName =
+  process.env['VULCAN_CACHE_LAYER_PERSISTENT_FILE_NAME'] ||
+  'vulcan_caching_layer.db';
+
 // The schema name for vulcan used to create table when loading cache files to cache data source
 export const vulcanCacheSchemaName = 'vulcan';
 
diff --git a/packages/core/test/cache-layer/cacheLayerLoader.spec.ts b/packages/core/test/cache-layer/cacheLayerLoader.spec.ts
@@ -1,5 +1,7 @@
 import * as fs from 'fs';
+import * as duckdb from 'duckdb';
 import * as sinon from 'ts-sinon';
+import * as path from 'path';
 import {
   CacheLayerInfo,
   CacheLayerLoader,
@@ -9,7 +11,7 @@ import {
   vulcanCacheSchemaName,
 } from '@vulcan-sql/core';
 import { MockDataSource, getQueryResults } from './mockDataSource';
-
+const db = new duckdb.Database(':memory:');
 describe('Test cache layer loader', () => {
   const folderPath = 'loader-test-exported-parquets';
   const profiles = [
@@ -108,4 +110,80 @@ describe('Test cache layer loader', () => {
     // Set 50s timeout to test cache loader export and load data
     50 * 10000
   );
+  it.each([
+    {
+      templateName: 'template-1',
+      cache: {
+        cacheTableName: 'employees',
+        sql: sinon.default.stub() as any,
+        profile: profiles[0].name,
+        folderSubpath: '2023',
+      } as CacheLayerInfo,
+    },
+    {
+      templateName: 'template-1',
+      cache: {
+        cacheTableName: 'departments',
+        sql: sinon.default.stub() as any,
+        profile: profiles[1].name,
+        folderSubpath: '2023',
+      } as CacheLayerInfo,
+    },
+  ])(
+    'Should use existed parquet to load cache table: $cache.cacheTableName',
+    async ({ templateName, cache }) => {
+      // Arrange
+      const { profile, cacheTableName, folderSubpath } = cache;
+      const dir = path.resolve(
+        folderPath,
+        templateName,
+        profile,
+        cacheTableName,
+        folderSubpath!
+      );
+      await createParquetFile(dir, cacheTableName);
+      // Act
+      const loader = new CacheLayerLoader(options, stubFactory as any);
+      await loader.load(templateName, cache);
+
+      // Assert
+      const actual = (
+        await getQueryResults(
+          "select * from information_schema.tables where table_schema = 'vulcan'"
+        )
+      ).map((row) => {
+        return {
+          table: row['table_name'],
+          schema: row['table_schema'],
+        };
+      });
+      expect(actual).toEqual(
+        expect.arrayContaining([
+          {
+            table: cache.cacheTableName,
+            schema: vulcanCacheSchemaName,
+          },
+        ])
+      );
+    },
+    // Set 50s timeout to test cache loader export and load data
+    50 * 10000
+  );
 });
+
+async function createParquetFile(path: string, fileName: string) {
+  if (!fs.existsSync(path)) {
+    fs.mkdirSync(path, { recursive: true });
+  }
+  db.run(`CREATE OR REPLACE TABLE parquet_table (i integer)`);
+  db.run(`INSERT INTO parquet_table (i) VALUES (1)`);
+  return new Promise((resolve, reject) => {
+    db.run(
+      `COPY (SELECT * FROM parquet_table) TO '${path}/${fileName}.parquet' (FORMAT PARQUET);`,
+      (err: any) => {
+        if (err) reject(err);
+        resolve(true);
+      }
+    );
+  });
+}
diff --git a/packages/core/test/containers/cacheLayer.spec.ts b/packages/core/test/containers/cacheLayer.spec.ts
@@ -1,6 +1,7 @@
 import {
   CacheLayerLoader,
   cacheLayerModule,
+  cacheLayerPersistentFileName,
   CacheLayerStoreFormatType,
   cacheProfileName,
   DataResult,
@@ -89,11 +90,21 @@ it('Cache layer module should add "vulcan.cache" profile and bind "duckdb" type
 
   expect(dsFromTestDuck.injectedProfiles).toEqual([
     { name: 'test-duck', type: 'duckdb', allow: '*' },
-    { name: cacheProfileName, type: 'duckdb', allow: '*' },
+    {
+      name: cacheProfileName,
+      type: 'duckdb',
+      allow: '*',
+      connection: { ['persistent-path']: cacheLayerPersistentFileName },
+    },
   ]);
   expect(dsFromCacheLayer.injectedProfiles).toEqual([
     { name: 'test-duck', type: 'duckdb', allow: '*' },
-    { name: cacheProfileName, type: 'duckdb', allow: '*' },
+    {
+      name: cacheProfileName,
+      type: 'duckdb',
+      allow: '*',
+      connection: { ['persistent-path']: cacheLayerPersistentFileName },
+    },
   ]);
 });
 

Original file line number	Diff line number	Diff line change
`@@ -114,6 +114,8 @@ export class CacheLayerInfo {`
`114`	`114`	`refreshExpression?: RefreshExpression;`
`115`	`115`	`// index key name -> index column`
`116`	`116`	`indexes?: Record<string, string>;`
	`117`	`+ // cache folder subpath`
	`118`	`+ folderSubpath?: string;`
`117`	`119`	`}`
`118`	`120`
`119`	`121`	`export class APISchema {`