fix(schema-compiler): Fix pre-agg partitions creation for Ksql

packages/cubejs-ksql-driver/src/KsqlQuery.ts

@@ -83,6 +83,10 @@ export class KsqlQuery extends BaseQuery {
     return dimensionColumns.length ? ` GROUP BY ${dimensionColumns.join(', ')}` : '';
   }
 
+  public partitionInvalidateKeyQueries(_cube: string, _preAggregation: any) {
+    return [];
+  }
+
   public preAggregationStartEndQueries(cube: string, preAggregation: any) {
     if (preAggregation.partitionGranularity) {
       if (!preAggregation.refreshRangeStart) {

packages/cubejs-schema-compiler/src/adapter/BaseQuery.js

@@ -4456,6 +4456,10 @@ export class BaseQuery {
     return `SELECT ${sql} as refresh_key`;
   }
 
+  partitionInvalidateKeyQueries(_cube, _preAggregation) {
+    // this is not used across all dialects, atm only in KsqlQuery.
+  }
+
   preAggregationInvalidateKeyQueries(cube, preAggregation, preAggregationName) {
     return this.cacheValue(
       ['preAggregationInvalidateKeyQueries', cube, JSON.stringify(preAggregation)],

packages/cubejs-schema-compiler/src/adapter/PreAggregations.ts

@@ -237,6 +237,8 @@ export class PreAggregations {
     const tableName = this.preAggregationTableName(cube, preAggregationName, preAggregation);
     const invalidateKeyQueries = this.query.preAggregationInvalidateKeyQueries(cube, preAggregation, preAggregationName);
     const queryForSqlEvaluation = this.query.preAggregationQueryForSqlEvaluation(cube, preAggregation);
+    // Atm this is only defined in KsqlQuery but without it partitions are recreated on every refresh
+    const partitionInvalidateKeyQueries = queryForSqlEvaluation.partitionInvalidateKeyQueries?.(cube, preAggregation);
 
     const allBackAliasMembers = this.query.allBackAliasMembers();
 
@@ -298,6 +300,7 @@ export class PreAggregations {
       timestampPrecision: queryForSqlEvaluation.timestampPrecision(),
       tableName,
       invalidateKeyQueries,
+      partitionInvalidateKeyQueries,
       type: preAggregation.type,
       external: preAggregation.external,
       previewSql: queryForSqlEvaluation.preAggregationPreviewSql(tableName),