Repository: dbt-labs/dbt-bigquery (language: python)

Issue:
[Bug] List types don't save to BigQuery using 1.7 (but did in 1.5)
### Is this a new bug in dbt-core?

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

### Current Behavior

Using the following code:
```python
import pandas as pd

def model(dbt, session):
    dbt.config(
        submission_method="cluster",
        dataproc_cluster_name="dbt-python",
        packages=["pandas", "numpy"]
    )

    # Define schema for the DataFrame
    schema = "id INT, embedding ARRAY<INT>"

    # Create a DataFrame
    data = [(5, [1, 2, 3]), (6, [4, 5, 6]), (7, [7, 8, 9])]
    df = session.createDataFrame(data, schema=schema)

    return df
```

The database result is:
![image (7)](https://github.com/dbt-labs/dbt-core/assets/80682772/47210b7d-b3d7-4197-ba15-ded74ae849c1)

As you can see the `embedding` column has no data


### Expected Behavior

Running the same Python code:
```
import pandas as pd

def model(dbt, session):
    dbt.config(
        submission_method="cluster",
        dataproc_cluster_name="dbt-python",
        packages=["pandas", "numpy"]
    )

    # Define schema for the DataFrame
    schema = "id INT, embedding ARRAY<INT>"

    # Create a DataFrame
    data = [(5, [1, 2, 3]), (6, [4, 5, 6]), (7, [7, 8, 9])]
    df = session.createDataFrame(data, schema=schema)

    return df
```

Should result in the following result in the target table in the data warehouse:
![image (8)](https://github.com/dbt-labs/dbt-core/assets/80682772/46148f2d-334a-4459-9ead-999635f2c9ce)


### Steps To Reproduce

NOTE: This worked without issue in dbt 1.5.9 and loaded the data correctly.

1. Using dbt-core 1.7.8 with BigQuery adapter 1.7.4 running with a DataProc cluster in GCP
2. Create a new dbt python model using the code described in the behaviour section
3. Run the model
4. Observe the results in the target table

### Relevant log output

_No response_

### Environment

```markdown
- OS: Mac OS 14 (Dataproc OS - 2.1.40-debian11)
- Python: 3.9.9
- dbt: 1.7.8
- bigquery-connector-version: 1.2.0
- spark-bigquery-connector-version: 0.36.1
```


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

bigquery

### Additional Context

As stated previously, this worked without issue on dbt 1.5.9 and loaded the data as expected. After updating to dbt 1.7.8 this has now broken and the data is no longer loaded correctly.


Discussion / hints:
We ran into the same issue with the version update.
Adding this in the `def model(dbt, session):` method fixed it for us. It's definitely a regression though.

```
session.conf.set('intermediateFormat', "orc")
``` 

https://github.com/dbt-labs/dbt-bigquery/issues/1047

Fix patch:
diff --git a/.changes/unreleased/Features-20240426-105319.yaml b/.changes/unreleased/Features-20240426-105319.yaml
new file mode 100644
index 000000000..0af2f9aa8
--- /dev/null
+++ b/.changes/unreleased/Features-20240426-105319.yaml
@@ -0,0 +1,7 @@
+kind: Features
+body: Add configuration options `enable_list_inference` and `intermediate_format` for python
+  models
+time: 2024-04-26T10:53:19.874239-04:00
+custom:
+  Author: mikealfare
+  Issue: 1047 1114
diff --git a/.changes/unreleased/Fixes-20240426-105224.yaml b/.changes/unreleased/Fixes-20240426-105224.yaml
new file mode 100644
index 000000000..624006ba5
--- /dev/null
+++ b/.changes/unreleased/Fixes-20240426-105224.yaml
@@ -0,0 +1,7 @@
+kind: Fixes
+body: Default `enableListInference` to `True` for python models to support nested
+  lists
+time: 2024-04-26T10:52:24.827314-04:00
+custom:
+  Author: mikealfare
+  Issue: 1047 1114
diff --git a/dbt/adapters/bigquery/impl.py b/dbt/adapters/bigquery/impl.py
index 2df35bc65..700c03719 100644
--- a/dbt/adapters/bigquery/impl.py
+++ b/dbt/adapters/bigquery/impl.py
@@ -99,6 +99,8 @@ class BigqueryConfig(AdapterConfig):
     enable_refresh: Optional[bool] = None
     refresh_interval_minutes: Optional[int] = None
     max_staleness: Optional[str] = None
+    enable_list_inference: Optional[bool] = None
+    intermediate_format: Optional[str] = None
 
 
 class BigQueryAdapter(BaseAdapter):
diff --git a/dbt/include/bigquery/macros/materializations/table.sql b/dbt/include/bigquery/macros/materializations/table.sql
index 68117b06a..e3c5b3598 100644
--- a/dbt/include/bigquery/macros/materializations/table.sql
+++ b/dbt/include/bigquery/macros/materializations/table.sql
@@ -49,12 +49,19 @@
 from pyspark.sql import SparkSession
 {%- set raw_partition_by = config.get('partition_by', none) -%}
 {%- set raw_cluster_by = config.get('cluster_by', none) -%}
+{%- set enable_list_inference = config.get('enable_list_inference', true) -%}
+{%- set intermediate_format = config.get('intermediate_format', none) -%}
+
 {%- set partition_config = adapter.parse_partition_by(raw_partition_by) %}
 
 spark = SparkSession.builder.appName('smallTest').getOrCreate()
 
 spark.conf.set("viewsEnabled","true")
 spark.conf.set("temporaryGcsBucket","{{target.gcs_bucket}}")
+spark.conf.set("enableListInference", "{{ enable_list_inference }}")
+{% if intermediate_format %}
+spark.conf.set("intermediateFormat", "{{ intermediate_format }}")
+{% endif %}
 
 {{ compiled_code }}
 dbt = dbtObj(spark.read.format("bigquery").load)


Test patch:
diff --git a/tests/functional/python_model_tests/__init__.py b/tests/functional/python_model_tests/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/functional/python_model_tests/files.py b/tests/functional/python_model_tests/files.py
new file mode 100644
index 000000000..1cb95602a
--- /dev/null
+++ b/tests/functional/python_model_tests/files.py
@@ -0,0 +1,125 @@
+SINGLE_RECORD = """
+import pandas as pd
+
+def model(dbt, session):
+
+    dbt.config(
+        submission_method="serverless",
+        materialized="table"
+    )
+
+    df = pd.DataFrame(
+        [
+            {"column_name": {"name": "hello", "my_list": ["h", "e", "l", "l", "o"]}},
+        ]
+    )
+
+    return df
+"""
+
+
+MULTI_RECORD = """
+import pandas as pd
+
+def model(dbt, session):
+
+    dbt.config(
+        submission_method="serverless",
+        materialized="table",
+    )
+
+    df = pd.DataFrame(
+        [
+            {"column_name": [{"name": "hello", "my_list": ["h", "e", "l", "l", "o"]}]},
+        ]
+    )
+
+    return df
+"""
+
+
+ORC_FORMAT = """
+import pandas as pd
+
+def model(dbt, session):
+
+    dbt.config(
+        submission_method="serverless",
+        materialized="table",
+        intermediate_format="orc",
+    )
+
+    df = pd.DataFrame(
+        [
+            {"column_name": [{"name": "hello", "my_list": ["h", "e", "l", "l", "o"]}]},
+        ]
+    )
+
+    return df
+"""
+
+
+ENABLE_LIST_INFERENCE = """
+import pandas as pd
+
+def model(dbt, session):
+
+    dbt.config(
+        submission_method="serverless",
+        materialized="table",
+        enable_list_inference="true",
+    )
+
+    df = pd.DataFrame(
+        [
+            {"column_name": [{"name": "hello", "my_list": ["h", "e", "l", "l", "o"]}]},
+        ]
+    )
+
+    return df
+"""
+
+
+ENABLE_LIST_INFERENCE_PARQUET_FORMAT = """
+import pandas as pd
+
+def model(dbt, session):
+
+    dbt.config(
+        submission_method="serverless",
+        materialized="table",
+        enable_list_inference="true",
+        intermediate_format="parquet",
+    )
+
+    df = pd.DataFrame(
+        [
+            {"column_name": [{"name": "hello", "my_list": ["h", "e", "l", "l", "o"]}]},
+        ]
+    )
+
+    return df
+"""
+
+
+DISABLE_LIST_INFERENCE_ORC_FORMAT = """
+import pandas as pd
+
+def model(dbt, session):
+
+    dbt.config(
+        submission_method="serverless",
+        materialized="table",
+        enable_list_inference="false",
+        intermediate_format="orc",
+    )
+
+    df = pd.DataFrame(
+        [
+            {"column_name": [{"name": "hello", "my_list": ["h", "e", "l", "l", "o"]}]},
+        ]
+    )
+
+    return df
+
+"""
diff --git a/tests/functional/python_model_tests/test_list_inference.py b/tests/functional/python_model_tests/test_list_inference.py
new file mode 100644
index 000000000..581e2010b
--- /dev/null
+++ b/tests/functional/python_model_tests/test_list_inference.py
@@ -0,0 +1,36 @@
+"""
+This test case addresses this regression: https://github.com/dbt-labs/dbt-bigquery/issues/1047
+
+As the comments point out, the issue appears when the settings are:
+    - list inference: off
+    - intermediate format: parquet
+
+Adjusting either of these alleviates the issue.
+
+When the regression was first reported, `files.MULTI_RECORD` failed while the other models passed.
+"""
+from dbt.tests.util import run_dbt_and_capture
+import pytest
+
+from tests.functional.python_model_tests import files
+
+
+class TestListInference:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            # this is what worked prior to this issue
+            "single_record.py": files.SINGLE_RECORD,
+            # this is the model that initially failed for this issue
+            "multi_record.py": files.MULTI_RECORD,
+            # these are explicit versions of the default settings
+            "enable_list_inference.py": files.ENABLE_LIST_INFERENCE,
+            "enable_list_inference_parquet_format.py": files.ENABLE_LIST_INFERENCE_PARQUET_FORMAT,
+            # orc format also resolves the issue, regardless of list inference
+            "orc_format.py": files.ORC_FORMAT,
+            "disable_list_inference_orc_format.py": files.DISABLE_LIST_INFERENCE_ORC_FORMAT,
+        }
+
+    def test_models_success(self, project, models):
+        result, output = run_dbt_and_capture(["run"])
+        assert len(result) == len(models)