Add option to specify executeProject

Author: shydefoo

python/batch-predictor/merlinpyspark/config.py

@@ -140,6 +140,11 @@ def table(self) -> str:
 
     def project(self) -> str:
         return self._project
+
+    def execute_project(self) -> str:
+        if self.options() is not None and "execute_project" in self.options():
+            return self.options()["execute_project"]
+        return ""
 
     def schema(self) -> str:
         return self._schema
@@ -246,6 +251,11 @@ def result_column(self) -> str:
     def options(self) -> MutableMapping[str, str]:
         return self._proto.options
 
+    def execute_project(self) -> str:
+        if self.options() is not None and "execute_project" in self.options():
+            return self.options()["execute_project"]
+        return ""
+
     def save_mode(self) -> str:
         return SaveMode.Name(self._proto.save_mode).lower()
 

python/batch-predictor/merlinpyspark/sink.py

@@ -73,7 +73,10 @@ def __init__(self, spark_session: SparkSession, config: MaxComputeSinkConfig):
         self._spark = spark_session
 
     def get_jdbc_url(self):
-        return f"jdbc:odps:{self._config.endpoint()}?project={self._config.project()}&accessId={self.get_access_id()}&accessKey={self.get_access_key()}&interactiveMode={self.get_interactive_mode()}&odpsNamespaceSchema=true&schema={self._config.schema()}&enableLimit=false"
+        url = f"jdbc:odps:{self._config.endpoint()}?project={self._config.project()}&accessId={self.get_access_id()}&accessKey={self.get_access_key()}&interactiveMode={self.get_interactive_mode()}&odpsNamespaceSchema=true&schema={self._config.schema()}&enableLimit=false&alwaysFallback=true&enableOdpsLogger=true"
+        if self._config.execute_project() is not None:
+            url += f"&executeProject={self._config.execute_project()}"
+        return url
 
     def get_query_timeout(self):
         return self._config.options().get("queryTimeout", "300")

python/batch-predictor/merlinpyspark/source.py

@@ -154,7 +154,10 @@ def features(self) -> Iterable[str]:
         return self._config.features()
 
     def get_jdbc_url(self):
-        return f"jdbc:odps:{self._config.endpoint()}?project={self._config.project()}&accessId={self.get_access_id()}&accessKey={self.get_access_key()}&interactiveMode={self.get_interactive_mode()}&odpsNamespaceSchema=true&schema={self._config.schema()}&enableLimit=false&autoSelectLimit={self.auto_select_limit}&alwaysFallback=true&enableOdpsLogger=true"
+        url = f"jdbc:odps:{self._config.endpoint()}?project={self._config.project()}&accessId={self.get_access_id()}&accessKey={self.get_access_key()}&interactiveMode={self.get_interactive_mode()}&odpsNamespaceSchema=true&schema={self._config.schema()}&enableLimit=false&autoSelectLimit={self.auto_select_limit}&alwaysFallback=true&enableOdpsLogger=true"
+        if self._config.execute_project() is not None:
+            url += f"&executeProject={self._config.execute_project()}"
+        return url
 
     def get_query_timeout(self):
         return self._config.options().get("queryTimeout", "300")