[Adapters][Salesforce] another set of improvements (#5694)

* alias database to data_space in project configs

* Update dbt_project.yml: example usage for data_space config
Update examples to showcase data tests support

* add data_transform_run_timeout in profiles.yml

* allows configuring statement options from config;
update the profiles.yml as an example

* needs to provide a default, otherwise it causes issues downstream due to null yaml value is converted to string "null"

* include a select * example, but this won't work; see the e2e/main.go example in salesforce adbc

* prepare to use adapter.load_dataframe to support salesforce seed

* fix a syntax error

* disable this model

* bump the sf version

* nit

GitOrigin-RevId: 708429f6135f17a35d710e4a75ca4b346f355f0a

Author: xuliangs

crates/dbt-fusion-adapter/src/base_adapter.rs

@@ -423,9 +423,17 @@ pub trait BaseAdapter: fmt::Display + fmt::Debug + AdapterTyping + Send + Sync {
     ) -> Result<Value, MinijinjaError>;
 
     /// load_dataframe
-    fn load_dataframe(&self, _state: &State, _args: &[Value]) -> Result<Value, MinijinjaError> {
-        unimplemented!("only available with BigQuery adapter")
-    }
+    #[allow(clippy::too_many_arguments)]
+    fn load_dataframe(
+        &self,
+        _state: &State,
+        _database: &str,
+        _schema: &str,
+        _table_name: &str,
+        _agate_table: Arc<AgateTable>,
+        _file_path: &str,
+        _field_delimiter: &str,
+    ) -> Result<Value, MinijinjaError>;
 
     /// upload_file
     fn upload_file(&self, _state: &State, _args: &[Value]) -> Result<Value, MinijinjaError> {

crates/dbt-fusion-adapter/src/bridge_adapter.rs

@@ -1644,12 +1644,28 @@ impl BaseAdapter for BridgeAdapter {
     }
 
     #[tracing::instrument(skip(self, state), level = "trace")]
-    fn load_dataframe(&self, state: &State, args: &[Value]) -> Result<Value, MinijinjaError> {
+    fn load_dataframe(
+        &self,
+        state: &State,
+        database: &str,
+        schema: &str,
+        table_name: &str,
+        agate_table: Arc<AgateTable>,
+        file_path: &str,
+        field_delimiter: &str,
+    ) -> Result<Value, MinijinjaError> {
         let mut conn = self.borrow_tlocal_connection(node_id_from_state(state))?;
         let query_ctx = query_ctx_from_state(state)?.with_desc("load_dataframe");
-        let result = self
-            .typed_adapter
-            .load_dataframe(&query_ctx, conn.as_mut(), args)?;
+        let result = self.typed_adapter.load_dataframe(
+            &query_ctx,
+            conn.as_mut(),
+            database,
+            schema,
+            table_name,
+            agate_table,
+            file_path,
+            field_delimiter,
+        )?;
         Ok(result)
     }
 }

crates/dbt-fusion-adapter/src/funcs.rs

@@ -147,7 +147,45 @@ pub fn dispatch_adapter_calls(
         "update_columns" => adapter.update_columns(state, args),
         "update_table_description" => adapter.update_table_description(state, args),
         "alter_table_add_columns" => adapter.alter_table_add_columns(state, args),
-        "load_dataframe" => adapter.load_dataframe(state, args),
+        "load_dataframe" => {
+            let iter = ArgsIter::new(
+                name,
+                &[
+                    "database",
+                    "schema",
+                    "table_name",
+                    "field_path",
+                    "agate_table",
+                    "field_delimiter",
+                ],
+                args,
+            );
+            let database = iter.next_arg::<&str>()?;
+            let schema = iter.next_arg::<&str>()?;
+            let table_name = iter.next_arg::<&str>()?;
+            let file_path = iter.next_arg::<&str>()?;
+            let agate_table = iter
+                .next_arg::<&Value>()?
+                .downcast_object::<AgateTable>()
+                .ok_or_else(|| {
+                    MinijinjaError::new(
+                        MinijinjaErrorKind::InvalidOperation,
+                        "agate_table must be an agate.Table",
+                    )
+                })?;
+            let field_delimiter = iter.next_arg::<&str>()?;
+            iter.finish()?;
+
+            adapter.load_dataframe(
+                state,
+                database,
+                schema,
+                table_name,
+                agate_table,
+                file_path,
+                field_delimiter,
+            )
+        }
         "upload_file" => adapter.upload_file(state, args),
         "get_bq_table" => adapter.get_bq_table(state, args),
         "describe_relation" => adapter.describe_relation(state, args),

crates/dbt-fusion-adapter/src/parse/adapter.rs

@@ -592,7 +592,16 @@ impl BaseAdapter for ParseAdapter {
         Ok(none_value())
     }
 
-    fn load_dataframe(&self, _state: &State, _args: &[Value]) -> Result<Value, MinijinjaError> {
+    fn load_dataframe(
+        &self,
+        _state: &State,
+        _database: &str,
+        _schema: &str,
+        _table_name: &str,
+        _agate_table: Arc<AgateTable>,
+        _file_path: &str,
+        _field_delimiter: &str,
+    ) -> Result<Value, MinijinjaError> {
         Ok(none_value())
     }
 

crates/dbt-fusion-adapter/src/sql_engine.rs

@@ -4,6 +4,7 @@ use crate::base_adapter::{AdapterFactory, backend_of};
 use crate::config::AdapterConfig;
 use crate::errors::{AdapterError, AdapterErrorKind, AdapterResult};
 use crate::query_comment::{EMPTY_CONFIG, QueryCommentConfig};
+use crate::record_and_replay::{RecordEngine, ReplayEngine};
 use crate::sql_types::{NaiveTypeFormatterImpl, TypeFormatter};
 use crate::stmt_splitter::StmtSplitter;
 
@@ -32,8 +33,6 @@ use std::sync::RwLock;
 use std::sync::{Arc, LazyLock};
 use std::{thread, time::Duration};
 
-use super::record_and_replay::{RecordEngine, ReplayEngine};
-
 type Options = Vec<(String, OptionValue)>;
 
 /// Naive statement splitter used in the MockAdapter

crates/dbt-fusion-adapter/src/typed_adapter.rs

@@ -27,6 +27,7 @@ use dbt_schemas::schemas::project::ModelConfig;
 use dbt_schemas::schemas::relations::base::{BaseRelation, ComponentName};
 use dbt_schemas::schemas::relations::relation_configs::BaseRelationConfig;
 use dbt_schemas::schemas::{CommonAttributes, InternalDbtNodeAttributes};
+use dbt_xdbc::salesforce::DATA_TRANSFORM_RUN_TIMEOUT;
 use dbt_xdbc::{Connection, QueryCtx};
 use minijinja::{State, Value, args};
 
@@ -106,12 +107,34 @@ pub trait TypedBaseAdapter: fmt::Debug + Send + Sync + AdapterTyping {
         } else {
             engine.split_statements(&sql, dialect)
         };
-        let options = options
+
+        // Configure options
+        let mut options = options
             .unwrap_or_default()
             .into_iter()
             .map(|(key, value)| (key, OptionValue::String(value)))
             .collect::<Vec<_>>();
 
+        // Configure warehouse specific options
+        #[allow(clippy::single_match)]
+        match self.adapter_type() {
+            AdapterType::Salesforce => {
+                if let Some(timeout) = engine.config("data_transform_run_timeout") {
+                    let timeout = timeout.parse::<i64>().map_err(|e| {
+                        AdapterError::new(
+                            AdapterErrorKind::Configuration,
+                            format!("data_transform_run_timeout must be an integer string: {e}",),
+                        )
+                    })?;
+                    options.push((
+                        DATA_TRANSFORM_RUN_TIMEOUT.to_string(),
+                        OptionValue::Int(timeout),
+                    ));
+                }
+            }
+            _ => {}
+        }
+
         let mut last_batch = None;
         for statement in statements {
             last_batch = Some(execute_query_with_retry(
@@ -627,13 +650,19 @@ pub trait TypedBaseAdapter: fmt::Debug + Send + Sync + AdapterTyping {
     }
 
     /// load dataframe only used by bigquery adapter
+    #[allow(clippy::too_many_arguments)]
     fn load_dataframe(
         &self,
         _query_ctx: &QueryCtx,
         _conn: &'_ mut dyn Connection,
-        _args: &[Value],
+        _database: &str,
+        _schema: &str,
+        _table_name: &str,
+        _agate_table: Arc<AgateTable>,
+        _file_path: &str,
+        _field_delimiter: &str,
     ) -> AdapterResult<Value> {
-        unimplemented!("only available with BigQuery adapter")
+        unimplemented!("only available with BigQuery or Salesforce adapter")
     }
 
     /// alter_table_add_columns

crates/dbt-loader/src/dbt_macro_assets/dbt-salesforce/macros/adapters.sql

@@ -0,0 +1,3 @@
+{% macro salesforce__persist_docs(relation, model, for_relation, for_columns) -%}
+  -- noop
+{% endmacro %}

crates/dbt-loader/src/dbt_macro_assets/dbt-salesforce/macros/materializations/seed.sql

@@ -0,0 +1,21 @@
+
+{% macro salesforce__create_csv_table(model, agate_table) %}
+    -- noop
+{% endmacro %}
+
+{% macro salesforce__reset_csv_table(model, full_refresh, old_relation, agate_table) %}
+    {{ adapter.drop_relation(old_relation) }}
+{% endmacro %}
+
+{% macro salesforce__load_csv_rows(model, agate_table) %}
+    {%- set delimiter = model['config'].get('delimiter', ',') -%}
+    {# only comma delimiter is supported for salesforce #}
+    {{ adapter.load_dataframe(
+        model['database'],
+        model['schema'],
+        model['alias'],
+        model['project_root'] | string ~ model['original_file_path'] | string,
+        agate_table,
+        ',',
+    ) }}
+{% endmacro %}

crates/dbt-schemas/src/schemas/profiles.rs

@@ -213,7 +213,7 @@ impl DbConfig {
             ],
             DbConfig::Databricks(_) => &["host", "http_path", "schema"],
             // TODO: Salesforce connection keys
-            DbConfig::Salesforce(_) => &[],
+            DbConfig::Salesforce(_) => &["login_url", "database", "data_transform_run_timeout"],
             // TODO: Trino and Datafusion connection keys
             DbConfig::Trino(_) => &[],
             DbConfig::Datafusion(_) => &[],
@@ -736,12 +736,18 @@ pub struct SalesforceDbConfig {
     pub private_key_path: Option<PathBuf>,
     pub login_url: Option<String>,
     pub username: Option<String>,
+    #[serde(default = "default_data_transform_run_timeout")]
+    pub data_transform_run_timeout: Option<i64>,
 }
 
 fn default_salesforce_database() -> Option<String> {
     Some("default".to_string())
 }
 
+fn default_data_transform_run_timeout() -> Option<i64> {
+    Some(180000) // 3 mins
+}
+
 #[derive(Serialize, JsonSchema)]
 #[serde(untagged)]
 #[serde(rename_all = "snake_case")]

crates/dbt-schemas/src/schemas/project/configs/data_test_config.rs

@@ -25,7 +25,7 @@ use crate::schemas::serde::{StringOrArrayOfStrings, bool_or_string_bool, u64_or_
 pub struct ProjectDataTestConfig {
     #[serde(rename = "+alias")]
     pub alias: Option<String>,
-    #[serde(rename = "+database", alias = "+project")]
+    #[serde(rename = "+database", alias = "+project", alias = "+data_space")]
     pub database: Option<String>,
     #[serde(default, rename = "+enabled", deserialize_with = "bool_or_string_bool")]
     pub enabled: Option<bool>,