chore(ksql-driver): Select queries for ksql allowed only from Cube Store. In order to query ksql create pre-aggregation first if Kafka download isn't enabled -- remove unloadWithoutTempTable as it's not needed due to readOnly flag on rollups

Author: paveltiunov

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

@@ -363,7 +363,6 @@ export class KsqlDriver extends BaseDriver implements DriverInterface {
   public capabilities(): DriverCapabilities {
     return {
       streamingSource: true,
-      unloadWithoutTempTable: !!this.config.kafkaHost,
     };
   }
 }

packages/cubejs-testing/birdbox-fixtures/lambda/schema/RequestsNonReadOnly.js

@@ -0,0 +1,83 @@
+cube("RequestsNonReadOnly", {
+  sql: `select 1 as tenant_id, 1 as deployment_id, 'req-1' as request_id, (NOW() - INTERVAL '1 day')::timestamp as timestamp
+      UNION ALL
+      select 2 as tenant_id, 1 as deployment_id, 'req-2' as request_id, (NOW() - INTERVAL '2 day')::timestamp as timestamp
+  `,
+  data_source: "postgres",
+  measures: {
+    count: {
+      type: "count",
+    },
+  },
+  dimensions: {
+    tenant_id: {
+      sql: `tenant_id`,
+      type: "number",
+    },
+    request_id: {
+      sql: `request_id`,
+      type: "string",
+    },
+    timestamp: {
+      sql: `timestamp`,
+      type: "time",
+    },
+  },
+  pre_aggregations: {
+    batch_streaming_lambda: {
+      type: `rollup_lambda`,
+      rollups: [batch, RequestsNonReadOnlyStream.stream],
+    },
+
+    batch: {
+      external: true,
+      type: "rollup",
+      measures: [count],
+      dimensions: [tenant_id, request_id, timestamp],
+      granularity: "day",
+      time_dimension: RequestsNonReadOnly.timestamp,
+      partition_granularity: "day",
+      build_range_start: { sql: "SELECT NOW() - INTERVAL '10 day'" },
+      build_range_end: { sql: "SELECT NOW()" },
+    },
+  },
+});
+
+cube("RequestsNonReadOnlyStream", {
+  dataSource: "ksql",
+
+  sql: `SELECT * FROM REQUESTS`,
+
+  measures: {
+    count: {
+      type: "count",
+    },
+  },
+  dimensions: {
+    tenant_id: {
+      sql: `TENANT_ID`,
+      type: "number",
+    },
+    request_id: {
+      sql: `REQUEST_ID`,
+      type: "string",
+    },
+    timestamp: {
+      sql: `TIMESTAMP`,
+      type: "time",
+    },
+  },
+  preAggregations: {
+    stream: {
+      streamOffset: "earliest",
+      type: `rollup`,
+      measures: [count],
+      dimensions: [tenant_id, request_id, timestamp],
+      time_dimension: RequestsNonReadOnlyStream.timestamp,
+      granularity: "day",
+      partition_granularity: "day",
+      build_range_start: { sql: "SELECT DATE_SUB(NOW(), interval '96 hour')" },
+      build_range_end: { sql: "SELECT NOW()" }
+    },
+  },
+});

packages/cubejs-testing/test/smoke-lambda.test.ts

@@ -116,6 +116,35 @@ describe('lambda', () => {
     expect(response.loadResponse.results[0].data.length).toEqual(3);
   });
 
+  test('Query lambda with ksql with create table ', async () => {
+    const query: Query = {
+      measures: ['RequestsNonReadOnly.count'],
+      dimensions: ['RequestsNonReadOnly.tenant_id', 'RequestsNonReadOnly.request_id'],
+      timeDimensions: [
+        {
+          dimension: 'RequestsNonReadOnly.timestamp',
+          granularity: 'day'
+        }
+      ],
+    };
+    // First call to trigger the pre-aggregation build
+    await client.load(query);
+    // We have to wait for cubestore to consume the data from Kafka. There is no way to know when it's done right now.
+    await pausePromise(5000);
+
+    const response = await client.load(query);
+
+    // @ts-ignore
+    expect(response.loadResponse.results[0].data.map(i => i['RequestsNonReadOnly.request_id'])).toEqual([
+      'req-2',
+      'req-1',
+      'req-stream-2'
+    ]);
+
+    // @ts-ignore
+    expect(response.loadResponse.results[0].data.length).toEqual(3);
+  });
+
   test('query', async () => {
     const query: Query = {
       measures: ['Orders.count'],