Feat: Adding support for Salesforce CDP query engine

Dockerfile

@@ -24,6 +24,11 @@ RUN npm i -g [email protected] \
     && yarn config set cache-folder /mnt/yarn-cache/cache \
     && yarn config set yarn-offline-mirror /mnt/yarn-offline-mirror
 
+# Install OpenJDK-11
+RUN apt-get update && \
+    apt-get install --assume-yes openjdk-11-jre-headless && \
+    apt-get clean;
+
 WORKDIR /opt/querybook
 
 COPY requirements requirements/
@@ -38,10 +43,11 @@ RUN pip install -r requirements/base.txt \
 COPY package.json yarn.lock ./
 RUN yarn install --pure-lockfile && npm rebuild node-sass
 
-
 # Copy everything else
 COPY . .
 
+RUN wget -P /opt/querybook/ https://github.com/forcedotcom/Salesforce-CDP-jdbc/releases/download/release_2021.10A/Salesforce-CDP-jdbc-1.10.0-java8.jar
+
 # Webpack if prod
 RUN if [ "${PRODUCTION}" = "true" ] ; then ./node_modules/.bin/webpack --mode=production; fi
 

querybook/server/lib/query_executor/all_executors.py

@@ -7,6 +7,7 @@
 from .executors.sqlalchemy import SnowflakeQueryExecutor, GenericSqlAlchemyQueryExecutor
 from .executors.bigquery import BigQueryQueryExecutor
 from .executors.trino import TrinoQueryExecutor
+from .executors.salesforce_cdp import SalesforceCdpExecutor
 
 ALL_PLUGIN_EXECUTORS = import_plugin("executor_plugin", "ALL_PLUGIN_EXECUTORS", [])
 
@@ -18,6 +19,7 @@
     GenericSqlAlchemyQueryExecutor,
     SnowflakeQueryExecutor,
     TrinoQueryExecutor,
+    SalesforceCdpExecutor,
 ] + ALL_PLUGIN_EXECUTORS
 
 

querybook/server/lib/query_executor/clients/salesforce_cdp.py

@@ -0,0 +1,59 @@
+from clients.google_client import get_google_credentials
+from lib.query_executor.base_client import ClientBaseClass, CursorBaseClass
+from lib.utils.json import safe_loads
+import jaydebeapi
+
+class SalesforceCdpClient(ClientBaseClass):
+    def __init__(self, username=None, password=None, loginurl=None, *args, **kwargs):
+        connection_string = 'jdbc:queryService-jdbc:' + loginurl
+        props = [username, password]
+        self._conn = jaydebeapi.connect('com.salesforce.cdp.queryservice.QueryServiceDriver',
+                        connection_string,
+                        props,
+                        '/opt/querybook/Salesforce-CDP-jdbc-1.10.0-java8.jar')
+        super(SalesforceCdpClient, self).__init__()
+
+    def cursor(self) -> CursorBaseClass:
+        return SalesforceCdpCursor(cursor=self._conn.cursor())
+
+
+class SalesforceCdpCursor(CursorBaseClass):
+    def __init__(self, cursor):
+        self._cursor = cursor
+
+    def run(self, query):
+        self._cursor.execute(query)
+
+    def cancel(self):
+        # Can't cancel (yet)
+        pass
+
+    def poll(self):
+        # Query should immediately start to block after
+        # run, so when it gets to poll it is already
+        # finished
+        return True
+
+    def get_one_row(self):
+        return self._convert_row(self._cursor.fetchone())
+
+    def get_n_rows(self, n: int):
+        converted_records = []
+        if self._cursor:
+            records = self._cursor.fetchmany(size=n)
+            if records:
+                for r in records:
+                    converted_record = self._convert_row(r)
+                    converted_records.append(converted_record)
+        return converted_records
+
+    def get_columns(self):
+        columns = []
+        for c in self._cursor.description:
+            columns.append(c[0])
+        return columns
+
+    def _convert_row(self, row):
+        if row:
+            return list(row)
+        return None

querybook/server/lib/query_executor/executor_template/templates.py

@@ -88,3 +88,9 @@
         helper="The JSON string used to log in as service account. If not provided then **GOOGLE_CREDS** from settings will be used.",
     )
 )
+
+salesforce_cdp_template = StructFormField(
+    password=FormField(hidden=True,helper="The password for the CDP Salesforce org instance."),
+    username=FormField(regex="\\w+",helper="The user name for the CDP Salesforce org instance."),
+    loginurl=FormField(helper="Login url, example https://login.salesforce.com")
+)

querybook/server/lib/query_executor/executors/salesforce_cdp.py

@@ -0,0 +1,22 @@
+from const.query_execution import QueryExecutionErrorType
+from lib.query_executor.base_executor import QueryExecutorBaseClass
+from lib.query_executor.clients.salesforce_cdp import SalesforceCdpClient
+from lib.query_executor.executor_template.templates import salesforce_cdp_template
+
+
+class SalesforceCdpExecutor(QueryExecutorBaseClass):
+    @classmethod
+    def EXECUTOR_NAME(cls):
+        return "Salesforce CDP"
+
+    @classmethod
+    def EXECUTOR_LANGUAGE(cls):
+        return "Salesforce CDP"
+
+    @classmethod
+    def _get_client(cls, client_setting):
+        return SalesforceCdpClient(**client_setting)
+
+    @classmethod
+    def EXECUTOR_TEMPLATE(cls):
+        return salesforce_cdp_template

requirements/base.txt

@@ -83,3 +83,6 @@ gspread==3.6.0
 # LDAP
 python-ldap==3.3.1
 requests-oauthlib==1.0.0
+
+# Salesforce CDP
+jaydebeapi==1.2.3