support dynamic variables in Python models | [CT-1450] [Bug] Unable to create python model for use across BigQuery and Databricks due to submission_method var

[pgoslatara]

Is this a new bug in dbt-core?

• I believe this is a new bug in dbt-core
• I have searched the existing issues, and I could not find an existing issue for this bug

Current Behavior

I'm developing a package that includes a python model that runs on all three supported databases (BigQuery, Databricks, Snowflake). Running on BigQuery requires a "submission_method" config to be "cluster" to run on a dedicated Dataproc cluster. Issue is that by including this I can no longer run the same model on Databricks as I get the following error message:

Submission method cluster is not supported for current adapter

I believe the error is related to this line, although I'm not sure of the intention of this check.

Expected Behavior

A python model can run across all supported databases with any desired config values.

Steps To Reproduce

1. Clone the bigquery_support branch from the dbt-currency-exchange-rates repo.
2. Follow the Development section to set up your environment.
3. Run dbt deps.
4. Run dbt build -t bigquery. Succeeds.
5. Run dbt build -t snowflake. Succeeds.
6. Run dbt build -t databricks. Fails.

Relevant log output

(venv13) ⋊> ~/w/G/2/dbt-currency-exchange-rates on bigquery_support  dbt build -t bigquery                      17:07:16
16:07:24  Running with dbt=1.3.0
16:07:24  Unable to do partial parsing because config vars, config profile, or config target have changed
16:07:24  Unable to do partial parsing because profile has changed
16:07:24  Unable to do partial parsing because env vars used in profiles.yml have changed
16:07:24  Unable to do partial parsing because a project dependency has been added
16:07:25  Found 1 model, 1 test, 0 snapshots, 0 analyses, 556 macros, 0 operations, 0 seed files, 0 sources, 0 exposures, 0 metrics
16:07:25
16:07:26  Concurrency: 1 threads (target='bigquery')
16:07:26
16:07:26  1 of 2 START python incremental model dbt_base.dbt_currency_exchange_rates ..... [RUN]
16:08:21  1 of 2 OK created python incremental model dbt_base.dbt_currency_exchange_rates  [MERGE (1.4k rows, 328.5 KB processed) in 54.69s]
16:08:21  2 of 2 START test dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date  [RUN]
16:08:22  2 of 2 PASS dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date  [PASS in 1.42s]
16:08:22
16:08:22  Finished running 1 incremental model, 1 test in 0 hours 0 minutes and 56.82 seconds (56.82s).
16:08:22
16:08:22  Completed successfully
16:08:22
16:08:22  Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2

(venv13) ⋊> ~/w/G/2/dbt-currency-exchange-rates on bigquery_support  dbt build -t snowflake                     17:08:23
16:10:34  Running with dbt=1.3.0
16:10:34  Unable to do partial parsing because config vars, config profile, or config target have changed
16:10:34  Unable to do partial parsing because env vars used in profiles.yml have changed
16:10:34  Unable to do partial parsing because a project dependency has been added
16:10:35  Found 1 model, 1 test, 0 snapshots, 0 analyses, 540 macros, 0 operations, 0 seed files, 0 sources, 0 exposures, 0 metrics
16:10:35
16:10:37  Concurrency: 1 threads (target='snowflake')
16:10:37
16:10:37  1 of 2 START python incremental model dbt_currency_exchange_rates_from_profile_yml_base_base.dbt_currency_exchange_rates  [RUN]
16:11:05  1 of 2 OK created python incremental model dbt_currency_exchange_rates_from_profile_yml_base_base.dbt_currency_exchange_rates  [SUCCESS 1352 in 28.14s]
16:11:05  2 of 2 START test dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date  [RUN]
16:11:06  2 of 2 PASS dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date  [PASS in 1.46s]
16:11:06
16:11:06  Finished running 1 incremental model, 1 test in 0 hours 0 minutes and 31.47 seconds (31.47s).
16:11:06
16:11:06  Completed successfully
16:11:06
16:11:06  Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2

(venv13) ⋊> ~/w/G/2/dbt-currency-exchange-rates on bigquery_support  dbt build -t databricks                    17:11:07
16:11:14  Running with dbt=1.3.0
16:11:14  Unable to do partial parsing because config vars, config profile, or config target have changed
16:11:14  Unable to do partial parsing because env vars used in profiles.yml have changed
16:11:14  Unable to do partial parsing because a project dependency has been added
16:11:16  Found 1 model, 1 test, 0 snapshots, 0 analyses, 600 macros, 0 operations, 0 seed files, 0 sources, 0 exposures, 0 metrics
16:11:16
16:11:17  Concurrency: 8 threads (target='databricks')
16:11:17
16:11:17  1 of 2 START python incremental model dbt_currency_exchange_rates_base.dbt_currency_exchange_rates  [RUN]
16:11:17  Unhandled error while executing model.dbt_currency_exchange_rates.dbt_currency_exchange_rates
Submission method cluster is not supported for current adapter
16:11:17  1 of 2 ERROR creating python incremental model dbt_currency_exchange_rates_base.dbt_currency_exchange_rates  [ERROR in 0.08s]
16:11:17  2 of 2 SKIP test dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date  [SKIP]
16:11:17
16:11:17  Finished running 1 incremental model, 1 test in 0 hours 0 minutes and 1.91 seconds (1.91s).
16:11:17
16:11:17  Completed with 1 error and 0 warnings:
16:11:17
16:11:17  Submission method cluster is not supported for current adapter
16:11:17
16:11:17  Done. PASS=0 WARN=0 ERROR=1 SKIP=1 TOTAL=2

(venv13) ⋊> ~/w/G/2/dbt-currency-exchange-rates on bigquery_support  dbt -d build -t databricks                 17:11:18


============================== 2022-10-31 16:11:26.721179 | 95a17d9c-eee6-4d04-8f68-0e20604b574e ==============================
16:11:26.721197 [info ] [MainThread]: Running with dbt=1.3.0
16:11:26.721703 [debug] [MainThread]: running dbt with arguments {'debug': True, 'write_json': True, 'use_colors': True, 'printer_width': 80, 'version_check': True, 'partial_parse': True, 'static_parser': True, 'profiles_dir': '<REDACTED>/dbt-currency-exchange-rates', 'send_anonymous_usage_stats': True, 'event_buffer_size': 100000, 'quiet': False, 'no_print': False, 'target': 'databricks', 'indirect_selection': 'eager', 'resource_types': [], 'which': 'build', 'rpc_method': 'build'}
16:11:26.722142 [debug] [MainThread]: Tracking: tracking
16:11:26.726863 [debug] [MainThread]: Sending event: {'category': 'dbt', 'action': 'invocation', 'label': 'start', 'context': [<snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee60ed9ee0>, <snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee60ed9fd0>, <snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee60ed9f70>]}
16:11:26.798339 [debug] [MainThread]: Partial parsing enabled: 0 files deleted, 0 files added, 0 files changed.
16:11:26.798721 [debug] [MainThread]: Partial parsing enabled, no changes found, skipping parsing
16:11:26.806221 [debug] [MainThread]: Sending event: {'category': 'dbt', 'action': 'load_project', 'label': '95a17d9c-eee6-4d04-8f68-0e20604b574e', 'context': [<snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee8ced10d0>]}
16:11:26.817986 [debug] [MainThread]: Sending event: {'category': 'dbt', 'action': 'resource_counts', 'label': '95a17d9c-eee6-4d04-8f68-0e20604b574e', 'context': [<snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee8d024ee0>]}
16:11:26.818639 [info ] [MainThread]: Found 1 model, 1 test, 0 snapshots, 0 analyses, 600 macros, 0 operations, 0 seed files, 0 sources, 0 exposures, 0 metrics
16:11:26.819068 [debug] [MainThread]: Sending event: {'category': 'dbt', 'action': 'runnable_timing', 'label': '95a17d9c-eee6-4d04-8f68-0e20604b574e', 'context': [<snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee8d024c70>]}
16:11:26.820616 [info ] [MainThread]:
16:11:26.821468 [debug] [MainThread]: Acquiring new databricks connection "master"
16:11:26.823006 [debug] [ThreadPool]: Acquiring new databricks connection "list_schemas"
16:11:26.834712 [debug] [ThreadPool]: Using databricks connection "list_schemas"
16:11:26.835142 [debug] [ThreadPool]: On list_schemas: /* {"app": "dbt", "dbt_version": "1.3.0", "dbt_databricks_version": "1.3.0", "databricks_sql_connector_version": "2.1.0", "profile_name": "dbt_currency_exchange_rates", "target_name": "databricks", "connection_name": "list_schemas"} */

    show databases

16:11:26.835461 [debug] [ThreadPool]: Opening a new connection, currently in state init
16:11:27.120029 [debug] [ThreadPool]: SQL status: OK in 0.28 seconds
16:11:27.126441 [debug] [ThreadPool]: On list_schemas: Close
16:11:27.196046 [debug] [ThreadPool]: Acquiring new databricks connection "list_None_dbt_currency_exchange_rates_base"
16:11:27.203207 [debug] [ThreadPool]: Spark adapter: NotImplemented: add_begin_query
16:11:27.203513 [debug] [ThreadPool]: Using databricks connection "list_None_dbt_currency_exchange_rates_base"
16:11:27.203842 [debug] [ThreadPool]: On list_None_dbt_currency_exchange_rates_base: /* {"app": "dbt", "dbt_version": "1.3.0", "dbt_databricks_version": "1.3.0", "databricks_sql_connector_version": "2.1.0", "profile_name": "dbt_currency_exchange_rates", "target_name": "databricks", "connection_name": "list_None_dbt_currency_exchange_rates_base"} */
show table extended in dbt_currency_exchange_rates_base like '*'

16:11:27.204114 [debug] [ThreadPool]: Opening a new connection, currently in state closed
16:11:27.795852 [debug] [ThreadPool]: SQL status: OK in 0.59 seconds
16:11:27.799756 [debug] [ThreadPool]: On list_None_dbt_currency_exchange_rates_base: ROLLBACK
16:11:27.800324 [debug] [ThreadPool]: Databricks adapter: NotImplemented: rollback
16:11:27.800861 [debug] [ThreadPool]: On list_None_dbt_currency_exchange_rates_base: Close
16:11:27.859394 [debug] [MainThread]: Sending event: {'category': 'dbt', 'action': 'runnable_timing', 'label': '95a17d9c-eee6-4d04-8f68-0e20604b574e', 'context': [<snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee60ed9ee0>]}
16:11:27.859955 [debug] [MainThread]: Spark adapter: NotImplemented: add_begin_query
16:11:27.860284 [debug] [MainThread]: Spark adapter: NotImplemented: commit
16:11:27.861012 [info ] [MainThread]: Concurrency: 8 threads (target='databricks')
16:11:27.861328 [info ] [MainThread]:
16:11:27.868909 [debug] [Thread-1  ]: Began running node model.dbt_currency_exchange_rates.dbt_currency_exchange_rates
16:11:27.869451 [info ] [Thread-1  ]: 1 of 2 START python incremental model dbt_currency_exchange_rates_base.dbt_currency_exchange_rates  [RUN]
16:11:27.870175 [debug] [Thread-1  ]: Acquiring new databricks connection "model.dbt_currency_exchange_rates.dbt_currency_exchange_rates"
16:11:27.870518 [debug] [Thread-1  ]: Began compiling node model.dbt_currency_exchange_rates.dbt_currency_exchange_rates
16:11:27.870905 [debug] [Thread-1  ]: Compiling model.dbt_currency_exchange_rates.dbt_currency_exchange_rates
16:11:27.891708 [debug] [Thread-1  ]: Writing injected SQL for node "model.dbt_currency_exchange_rates.dbt_currency_exchange_rates"
16:11:27.892278 [debug] [Thread-1  ]: finished collecting timing info
16:11:27.892644 [debug] [Thread-1  ]: Began executing node model.dbt_currency_exchange_rates.dbt_currency_exchange_rates
16:11:27.939254 [debug] [Thread-1  ]: On model.dbt_currency_exchange_rates.dbt_currency_exchange_rates:


from datetime import datetime, timedelta
import json
from typing import Dict


def get_schema(profile_type: str):
    if profile_type in ("bigquery", "databricks"):
        from pyspark.sql.types import StructField, StructType, StringType, FloatType
    elif profile_type == "snowflake":
        from snowflake.snowpark.types import (
            StructField,
            StructType,
            StringType,
            FloatType,
        )

    return StructType(
        [
            StructField("date", StringType(), False),
            StructField("currency_code", StringType(), False),
            StructField("rate", FloatType(), False),
        ]
    )


def prep_dict_for_dataframe(data: Dict) -> Dict:
    rates = []
    for d in list(data["rates"].keys()):
        for k, v in data["rates"][d].items():
            rates.append({"date": d, "currency_code": k, "rate": float(v)})

    return rates


def model(dbt, session):
    profile_type = str(dbt.config.get("profile_type"))
    lookback_days = str(dbt.config.get("lookback_days"))
    starsnow_schema = str(dbt.config.get("starsnow_schema"))

    # if profile_type == "bigquery":
    #     dbt.config(
    #         submission_method="cluster"
    #     )

    dbt.config(
        materialized="incremental",
        packages=["requests"],
        unique_key=["currency_code", "date"],
        # submission_method="cluster"
    )

    if dbt.is_incremental:
        start = (datetime.utcnow() - timedelta(int(lookback_days))).strftime("%Y-%m-%d")
    else:
        start = datetime(2000, 1, 1).strftime("%Y-%m-%d")

    end = datetime.utcnow().strftime("%Y-%m-%d")

    if profile_type in ("bigquery", "databricks"):
        import requests

        r = requests.get(
            f"https://api.exchangerate.host/timeseries?start_date={start}&end_date={end}"
        )
        df = spark.createDataFrame(prep_dict_for_dataframe(r.json()), get_schema(profile_type))

    elif profile_type == "snowflake":
        from snowflake.snowpark.functions import call_udf, col

        # Snowpark does not support external http calls: https://community.snowflake.com/s/question/0D53r0000BeAAgHCQW/error-in-calling-rest-api-endpoint-using-requests-get-post-method-in-snowpark-python-stored-procedure
        # As a proof-of-concept, the STARSNOW_REQUEST function can be installed and used to perform http calls: https://github.com/starschema/starsnow_request#deploying

        starsnow_params = session.create_dataframe(
            [
                [
                    f"https://api.exchangerate.host/timeseries?start_date={start}&end_date={end}",
                    {
                        "method": "get",
                    },
                ]
            ],
            schema=["url", "params"],
        )

        df = starsnow_params.select(
            call_udf(
                f"{starsnow_schema}.STARSNOW_REQUEST",
                col("url"),
                col("params"),
            ).as_("response")
        )
        rates_dict = json.loads(
            df.select("response").collect()[0].as_dict()["RESPONSE"]
        )["data"]

        df = session.create_dataframe(
            prep_dict_for_dataframe(rates_dict), get_schema(profile_type)
        )

    else:
        raise ValueError("Only BigQuery, Databricks and Snowflake are supported.")

    return df


# This part is user provided model code
# you will need to copy the next section to run the code
# COMMAND ----------
# this part is dbt logic for get ref work, do not modify

def ref(*args,dbt_load_df_function):
    refs = {}
    key = ".".join(args)
    return dbt_load_df_function(refs[key])


def source(*args, dbt_load_df_function):
    sources = {}
    key = ".".join(args)
    return dbt_load_df_function(sources[key])


config_dict = {'profile_type': 'databricks', 'lookback_days': 7, 'starsnow_schema': 'PUBLIC'}


class config:
    def __init__(self, *args, **kwargs):
        pass

    @staticmethod
    def get(key, default=None):
        return config_dict.get(key, default)

class this:
    """dbt.this() or dbt.this.identifier"""
    database = 'None'
    schema = 'dbt_currency_exchange_rates_base'
    identifier = 'dbt_currency_exchange_rates'
    def __repr__(self):
        return 'dbt_currency_exchange_rates_base.dbt_currency_exchange_rates'


class dbtObj:
    def __init__(self, load_df_function) -> None:
        self.source = lambda *args: source(*args, dbt_load_df_function=load_df_function)
        self.ref = lambda *args: ref(*args, dbt_load_df_function=load_df_function)
        self.config = config
        self.this = this()
        self.is_incremental = True

# COMMAND ----------

# how to execute python model in notebook
# dbt = dbtObj(spark.table)
# df = model(dbt, spark)


# --- Autogenerated dbt materialization code. --- #
dbt = dbtObj(spark.table)
df = model(dbt, spark)

# make sure pyspark exists in the namepace, for 7.3.x-scala2.12 it does not exist
import pyspark
# make sure pandas exists before using it
try:
  import pandas
  pandas_available = True
except ImportError:
  pandas_available = False

# make sure pyspark.pandas exists before using it
try:
  import pyspark.pandas
  pyspark_pandas_api_available = True
except ImportError:
  pyspark_pandas_api_available = False

# make sure databricks.koalas exists before using it
try:
  import databricks.koalas
  koalas_available = True
except ImportError:
  koalas_available = False

# preferentially convert pandas DataFrames to pandas-on-Spark or Koalas DataFrames first
# since they know how to convert pandas DataFrames better than `spark.createDataFrame(df)`
# and converting from pandas-on-Spark to Spark DataFrame has no overhead
if pyspark_pandas_api_available and pandas_available and isinstance(df, pandas.core.frame.DataFrame):
  df = pyspark.pandas.frame.DataFrame(df)
elif koalas_available and pandas_available and isinstance(df, pandas.core.frame.DataFrame):
  df = databricks.koalas.frame.DataFrame(df)

# convert to pyspark.sql.dataframe.DataFrame
if isinstance(df, pyspark.sql.dataframe.DataFrame):
  pass  # since it is already a Spark DataFrame
elif pyspark_pandas_api_available and isinstance(df, pyspark.pandas.frame.DataFrame):
  df = df.to_spark()
elif koalas_available and isinstance(df, databricks.koalas.frame.DataFrame):
  df = df.to_spark()
elif pandas_available and isinstance(df, pandas.core.frame.DataFrame):
  df = spark.createDataFrame(df)
else:
  msg = f"{type(df)} is not a supported type for dbt Python materialization"
  raise Exception(msg)

df.write.mode("overwrite").format("delta").option("overwriteSchema", "true").saveAsTable("dbt_currency_exchange_rates_base.dbt_currency_exchange_rates__dbt_tmp")

16:11:27.939728 [debug] [Thread-1  ]: finished collecting timing info
16:11:27.940499 [error] [Thread-1  ]: Unhandled error while executing model.dbt_currency_exchange_rates.dbt_currency_exchange_rates
Submission method cluster is not supported for current adapter
16:11:27.940848 [debug] [Thread-1  ]:
Traceback (most recent call last):
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/task/base.py", line 385, in safe_run
    result = self.compile_and_execute(manifest, ctx)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/task/base.py", line 338, in compile_and_execute
    result = self.run(ctx.node, manifest)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/task/base.py", line 429, in run
    return self.execute(compiled_node, manifest)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/task/run.py", line 281, in execute
    result = MacroGenerator(
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/clients/jinja.py", line 326, in __call__
    return self.call_macro(*args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/clients/jinja.py", line 253, in call_macro
    return macro(*args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/runtime.py", line 763, in __call__
    return self._invoke(arguments, autoescape)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/runtime.py", line 777, in _invoke
    rv = self._func(*arguments)
  File "<template>", line 126, in macro
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/sandbox.py", line 393, in call
    return __context.call(__obj, *args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/runtime.py", line 298, in call
    return __obj(*args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/clients/jinja.py", line 326, in __call__
    return self.call_macro(*args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/clients/jinja.py", line 253, in call_macro
    return macro(*args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/runtime.py", line 763, in __call__
    return self._invoke(arguments, autoescape)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/runtime.py", line 777, in _invoke
    rv = self._func(*arguments)
  File "<template>", line 55, in macro
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/sandbox.py", line 393, in call
    return __context.call(__obj, *args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/jinja2/runtime.py", line 298, in call
    return __obj(*args, **kwargs)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/context/providers.py", line 1263, in submit_python_job
    return self.adapter.submit_python_job(parsed_model, compiled_code)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/adapters/base/impl.py", line 141, in execution_with_log
    response = code_execution_function(*args)
  File "<REDACTED>/dbt-currency-exchange-rates/venv13/lib/python3.9/site-packages/dbt/adapters/base/impl.py", line 1229, in submit_python_job
    raise NotImplementedError(
NotImplementedError: Submission method cluster is not supported for current adapter
16:11:27.948505 [debug] [Thread-1  ]: Sending event: {'category': 'dbt', 'action': 'run_model', 'label': '95a17d9c-eee6-4d04-8f68-0e20604b574e', 'context': [<snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee8ce73ca0>]}
16:11:27.949177 [error] [Thread-1  ]: 1 of 2 ERROR creating python incremental model dbt_currency_exchange_rates_base.dbt_currency_exchange_rates  [ERROR in 0.08s]
16:11:27.949799 [debug] [Thread-1  ]: Finished running node model.dbt_currency_exchange_rates.dbt_currency_exchange_rates
16:11:27.951621 [debug] [Thread-3  ]: Began running node test.dbt_currency_exchange_rates.dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date.2349786b65
16:11:27.952111 [info ] [Thread-3  ]: 2 of 2 SKIP test dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date  [SKIP]
16:11:27.952582 [debug] [Thread-3  ]: Finished running node test.dbt_currency_exchange_rates.dbt_utils_unique_combination_of_columns_dbt_currency_exchange_rates_currency_code__date.2349786b65
16:11:27.955222 [debug] [MainThread]: Acquiring new databricks connection "master"
16:11:27.955642 [debug] [MainThread]: On master: ROLLBACK
16:11:27.956058 [debug] [MainThread]: Opening a new connection, currently in state init
16:11:28.024224 [debug] [MainThread]: Databricks adapter: NotImplemented: rollback
16:11:28.024594 [debug] [MainThread]: Spark adapter: NotImplemented: add_begin_query
16:11:28.024988 [debug] [MainThread]: Spark adapter: NotImplemented: commit
16:11:28.025426 [debug] [MainThread]: On master: ROLLBACK
16:11:28.025710 [debug] [MainThread]: Databricks adapter: NotImplemented: rollback
16:11:28.026065 [debug] [MainThread]: On master: Close
16:11:28.078860 [info ] [MainThread]:
16:11:28.079358 [info ] [MainThread]: Finished running 1 incremental model, 1 test in 0 hours 0 minutes and 1.26 seconds (1.26s).
16:11:28.080070 [debug] [MainThread]: Connection 'master' was properly closed.
16:11:28.080820 [debug] [MainThread]: Connection 'model.dbt_currency_exchange_rates.dbt_currency_exchange_rates' was properly closed.
16:11:28.096884 [info ] [MainThread]:
16:11:28.097390 [info ] [MainThread]: Completed with 1 error and 0 warnings:
16:11:28.097771 [info ] [MainThread]:
16:11:28.098185 [error] [MainThread]: Submission method cluster is not supported for current adapter
16:11:28.098579 [info ] [MainThread]:
16:11:28.098913 [info ] [MainThread]: Done. PASS=0 WARN=0 ERROR=1 SKIP=1 TOTAL=2
16:11:28.099367 [debug] [MainThread]: Sending event: {'category': 'dbt', 'action': 'invocation', 'label': 'end', 'context': [<snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee8ce60f40>, <snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee8ce60ee0>, <snowplow_tracker.self_describing_json.SelfDescribingJson object at 0x7fee8a536370>]}
16:11:28.099843 [debug] [MainThread]: Flushing usage events

Environment

- OS:

No LSB modules are available.
Distributor ID: Ubuntu
Description:    Ubuntu 20.04.5 LTS
Release:        20.04
Codename:       focal

• Python: Python 3.9.10
• dbt:

dbt-bigquery==1.3.0
dbt-core==1.3.0
dbt-databricks==1.3.0
dbt-snowflake==1.3.0

### Which database adapter are you using with dbt?

snowflake, bigquery, other (mention it in "Additional Context")

### Additional Context

dbt-databricks

[jtcohen6]

@pgoslatara Thanks for opening! (Sorry I pressed "close" by accident while typing!)

I see this code here in your linked demo (neat!):

    # if profile_type == "bigquery":
    #     dbt.config(
    #         submission_method="cluster"
    #     )

    dbt.config(
        materialized="incremental",
        packages=["requests"],
        unique_key=["currency_code", "date"],
        # submission_method="cluster"
    )

This is a rougher edge of Python models right now. dbt doesn't support more complex configuration (like ternary operators / conditional expressions) within Python code. Why? All parse-time attributes (ref, source, config, ...) are extracted via static analysis of Python code, not with an actual Python runtime. This allows us to preserve the clear separation between dbt-core as a Python app, and user-provided Python code that should execute completely in the data platform/engine.

Here's something you can do:

# models/config.yml
version: 2
models:
  - name: dbt_currency_exchange_rates
    config:
      submission_method: "{{ 'cluster' if target.type == 'bigquery' else 'all_purpose_cluster' }}"

---

Short of supporting actual execution of user Python code, I'm not sure how we should seek to support this in a more ergonomic way / with in-model code only.

[pgoslatara]

submission_method: "{{ 'cluster' if target.type == 'bigquery' else 'all_purpose_cluster' }}"

Nice workaround @jtcohen6!

[lostmygithubaccount]

converted to a discussion to keep this more searchable and crowd-source ideas on how we can go about fixing this. related issue: #5887

in short, how do we support dynamically generated strings and variables in Python models? we do not want to run user-space Python code within dbt