feat(cubesql): Data source per member (#9537)

This should fix SQL push-down targeting views.
Before this, views would always have a default datasource, and could use an incorrect dialect in multi-datasource deployments.

Now, the JS side would pass the "member -> data source" mapping to SQL API, and this would reflect each view member properly. Using members as keys is necessary for multi-data-source views (even if they are not supported right now in SQL pushdown, they are valid as a data model, and can be used with rollupJoin pre-aggregations)

Author: mcheshkov

packages/cubejs-api-gateway/src/gateway.ts

@@ -1486,15 +1486,14 @@ class ApiGateway {
       const query = {
         requestId: context.requestId,
       };
-      const cubeNameToDataSource = await compilerApi.cubeNameToDataSource(query);
+      const memberToDataSource: Record<string, string> = await compilerApi.memberToDataSource(query);
 
-      let dataSources = Object.keys(cubeNameToDataSource).map(c => cubeNameToDataSource[c]);
-      dataSources = [...new Set(dataSources)];
+      const dataSources = new Set(Object.values(memberToDataSource));
       const dataSourceToSqlGenerator = (await Promise.all(
-        dataSources.map(async dataSource => ({ [dataSource]: (await compilerApi.getSqlGenerator(query, dataSource)).sqlGenerator }))
+        [...dataSources].map(async dataSource => ({ [dataSource]: (await compilerApi.getSqlGenerator(query, dataSource)).sqlGenerator }))
       )).reduce((a, b) => ({ ...a, ...b }), {});
 
-      res({ cubeNameToDataSource, dataSourceToSqlGenerator });
+      res({ memberToDataSource, dataSourceToSqlGenerator });
     } catch (e: any) {
       this.handleError({
         e, context, res, requestStarted

packages/cubejs-backend-native/Cargo.lock

@@ -142,7 +142,7 @@ impl TransportService for NodeBridgeTransport {
 
         let channel = self.channel.clone();
 
-        let (cube_to_data_source, data_source_to_sql_generator) =
+        let (member_to_data_source, data_source_to_sql_generator) =
             call_raw_js_with_channel_as_callback(
                 self.channel.clone(),
                 self.sql_generators.clone(),
@@ -152,14 +152,14 @@ impl TransportService for NodeBridgeTransport {
                     let obj = v
                         .downcast::<JsObject, _>(cx)
                         .map_err(|e| CubeError::user(e.to_string()))?;
-                    let cube_to_data_source_obj = obj
-                        .get::<JsObject, _, _>(cx, "cubeNameToDataSource")
-                        .map_cube_err("Can't cast cubeNameToDataSource to object")?;
 
-                    let cube_to_data_source =
-                        key_to_values(cx, cube_to_data_source_obj, |cx, v| {
+                    let member_to_data_source_obj = obj
+                        .get::<JsObject, _, _>(cx, "memberToDataSource")
+                        .map_cube_err("Can't cast memberToDataSource to object")?;
+                    let member_to_data_source =
+                        key_to_values(cx, member_to_data_source_obj, |cx, v| {
                             let res = v.downcast::<JsString, _>(cx).map_cube_err(
-                                "Can't cast value to string in cube_to_data_source",
+                                "Can't cast value to string in member_to_data_source",
                             )?;
                             Ok(res.value(cx))
                         })?;
@@ -183,7 +183,7 @@ impl TransportService for NodeBridgeTransport {
                             Ok(res)
                         })?;
 
-                    Ok((cube_to_data_source, data_source_to_sql_generator))
+                    Ok((member_to_data_source, data_source_to_sql_generator))
                 }),
             )
             .await?;
@@ -208,7 +208,7 @@ impl TransportService for NodeBridgeTransport {
         })?;
         Ok(Arc::new(MetaContext::new(
             response.cubes.unwrap_or_default(),
-            cube_to_data_source,
+            member_to_data_source,
             data_source_to_sql_generator,
             compiler_id,
         )))

packages/cubejs-backend-native/src/transport.rs

@@ -98,6 +98,7 @@ const meta_fixture = require('./meta');
 
     return {
       cubeNameToDataSource: {},
+      memberToDataSource: {},
       dataSourceToSqlGenerator: {},
     };
   };

packages/cubejs-backend-native/test/server.js

@@ -100,6 +100,7 @@ function interfaceMethods() {
 
       return {
         cubeNameToDataSource: {},
+        memberToDataSource: {},
         dataSourceToSqlGenerator: {},
       };
     }),

packages/cubejs-backend-native/test/sql.test.ts

@@ -602,6 +602,34 @@ export class CompilerApi {
       ).reduce((a, b) => ({ ...a, ...b }), {});
   }
 
+  async memberToDataSource(query) {
+    const { cubeEvaluator } = await this.getCompilers({ requestId: query.requestId });
+
+    const entries = cubeEvaluator
+      .cubeNames()
+      .flatMap(cube => {
+        const cubeDef = cubeEvaluator.cubeFromPath(cube);
+        if (cubeDef.isView) {
+          const viewName = cubeDef.name;
+          return cubeDef.includedMembers.map(included => {
+            const memberName = `${viewName}.${included.name}`;
+            const refCubeDef = cubeEvaluator.cubeFromPath(included.memberPath);
+            const dataSource = refCubeDef.dataSource ?? 'default';
+            return [memberName, dataSource];
+          });
+        } else {
+          const cubeName = cubeDef.name;
+          const dataSource = cubeDef.dataSource ?? 'default';
+          return [
+            ...Object.keys(cubeDef.dimensions),
+            ...Object.keys(cubeDef.measures),
+            ...Object.keys(cubeDef.segments),
+          ].map(mem => [`${cubeName}.${mem}`, dataSource]);
+        }
+      });
+    return Object.fromEntries(entries);
+  }
+
   async dataSources(orchestratorApi, query) {
     const cubeNameToDataSource = await this.cubeNameToDataSource(query || { requestId: `datasources-${uuidv4()}` });
 

packages/cubejs-server-core/src/core/CompilerApi.js

@@ -50,3 +50,10 @@ cube(`Products`, {
 
   dataSource: 'products',
 });
+
+view(`ProductsView`, {
+  cubes: [{
+    joinPath: Products,
+    includes: `*`,
+  }]
+});

packages/cubejs-testing/birdbox-fixtures/multidb/schema/Products.js

@@ -32,3 +32,10 @@ cube(`Suppliers`, {
 
   dataSource: 'suppliers',
 });
+
+view(`SuppliersView`, {
+  cubes: [{
+    joinPath: Suppliers,
+    includes: `*`,
+  }]
+});

packages/cubejs-testing/birdbox-fixtures/multidb/schema/Suppliers.js

@@ -8,6 +8,14 @@ Array [
 ]
 `;
 
+exports[`multidb SQL pushdown queries to different data sources: ProductsView 1`] = `
+Array [
+  Object {
+    "name": "apples",
+  },
+]
+`;
+
 exports[`multidb SQL pushdown queries to different data sources: Suppliers 1`] = `
 Array [
   Object {
@@ -16,6 +24,14 @@ Array [
 ]
 `;
 
+exports[`multidb SQL pushdown queries to different data sources: SuppliersView 1`] = `
+Array [
+  Object {
+    "company": "Fruits Inc",
+  },
+]
+`;
+
 exports[`multidb query: query 1`] = `
 Array [
   Object {

packages/cubejs-testing/test/__snapshots__/smoke-multidb.test.ts.snap

@@ -120,6 +120,20 @@ describe('multidb', () => {
     expect(res.rows).toMatchSnapshot();
   });
 
+  test('SQL pushdown queries to different data sources: ProductsView', async () => {
+    const res = await connection.query(`
+  SELECT
+    name
+  FROM
+    ProductsView
+  WHERE
+    LOWER(name) = 'apples'
+  GROUP BY
+    1
+    `);
+    expect(res.rows).toMatchSnapshot();
+  });
+
   test('SQL pushdown queries to different data sources: Suppliers', async () => {
     const res = await connection.query(`
   SELECT
@@ -133,4 +147,18 @@ describe('multidb', () => {
     `);
     expect(res.rows).toMatchSnapshot();
   });
+
+  test('SQL pushdown queries to different data sources: SuppliersView', async () => {
+    const res = await connection.query(`
+  SELECT
+    company
+  FROM
+    SuppliersView
+  WHERE
+    LOWER(company) = 'fruits inc'
+  GROUP BY
+    1
+    `);
+    expect(res.rows).toMatchSnapshot();
+  });
 });