add option to partition large models (#939)

* add option to partition large models

* add explicit partition check

* undo dbt project change

* refactor: simplify BQ partitioning - enable by default, single disable flag

- Partitioning is now always on for BigQuery (partition by created_at/day)
  instead of requiring an opt-in flag
- Replace partition_run_results + run_results_partition_by vars with a
  single bigquery_disable_partitioning escape hatch
- Hardcode partition spec in model configs instead of indirecting through
  config vars
- Restore original code formatting (revert gratuitous style changes)
- Tighten test assertions to verify partition column is created_at

Note: existing BigQuery users will need --full-refresh to apply
partitioning to already-created tables.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: extract get_default_partition_by dispatched macro

Move the BQ partition logic into a dispatched macro to avoid repeating
the target.type check across model files. Follows the same pattern as
get_default_incremental_strategy and get_default_table_type.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Itamar Hartstein <haritamar@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 3 people

integration_tests/tests/test_dbt_artifacts/test_artifacts.py

@@ -163,3 +163,50 @@ def test_timings(dbt_project: DbtProject):
 
     assert len(results) == 1
     assert results[0]["execute_started_at"]
+
+
+@pytest.mark.only_on_targets(["bigquery"])
+def test_run_results_partitioned(dbt_project: DbtProject):
+    # BigQuery partitioning is enabled by default. Verify the model works and data is readable.
+    dbt_project.dbt_runner.vars["disable_run_results"] = False
+    # Full-refresh to ensure the table is created with partitioning
+    dbt_project.dbt_runner.run(select="dbt_run_results", full_refresh=True)
+    dbt_project.dbt_runner.run(select=TEST_MODEL)
+    results = dbt_project.run_query(
+        """select * from {{ ref("dbt_run_results") }} where name='%s'""" % TEST_MODEL
+    )
+    assert len(results) >= 1
+
+    # Verify the partition column is created_at in BigQuery
+    partition_cols = dbt_project.run_query(
+        "SELECT column_name "
+        "FROM `{{ ref('dbt_run_results').database }}.{{ ref('dbt_run_results').schema }}.INFORMATION_SCHEMA.COLUMNS` "
+        "WHERE table_name = '{{ ref('dbt_run_results').identifier }}' "
+        "AND is_partitioning_column = 'YES'"
+    )
+    assert [row["column_name"] for row in partition_cols] == [
+        "created_at"
+    ], "dbt_run_results should be partitioned by created_at in BigQuery"
+
+
+@pytest.mark.only_on_targets(["bigquery"])
+def test_dbt_invocations_partitioned(dbt_project: DbtProject):
+    # BigQuery partitioning is enabled by default. Verify dbt_invocations works.
+    dbt_project.dbt_runner.vars["disable_dbt_invocation_autoupload"] = False
+    # Full-refresh to ensure the table is created with partitioning
+    dbt_project.dbt_runner.run(select="dbt_invocations", full_refresh=True)
+    dbt_project.dbt_runner.run(selector="one")
+    dbt_project.read_table(
+        "dbt_invocations", where="yaml_selector = 'one'", raise_if_empty=True
+    )
+
+    # Verify the partition column is created_at in BigQuery
+    partition_cols = dbt_project.run_query(
+        "SELECT column_name "
+        "FROM `{{ ref('dbt_invocations').database }}.{{ ref('dbt_invocations').schema }}.INFORMATION_SCHEMA.COLUMNS` "
+        "WHERE table_name = '{{ ref('dbt_invocations').identifier }}' "
+        "AND is_partitioning_column = 'YES'"
+    )
+    assert [row["column_name"] for row in partition_cols] == [
+        "created_at"
+    ], "dbt_invocations should be partitioned by created_at in BigQuery"

macros/edr/system/system_utils/get_config_var.sql

@@ -143,6 +143,7 @@
         "anomaly_exclude_metrics": none,
         "disable_samples_on_pii_tags": false,
         "pii_tags": ["pii"],
+        "bigquery_disable_partitioning": false,
     } %}
     {{- return(default_config) -}}
 {%- endmacro -%}

macros/utils/cross_db_utils/partition_by.sql

@@ -0,0 +1,18 @@
+{% macro get_default_partition_by() %}
+    {% do return(adapter.dispatch("get_default_partition_by", "elementary")()) %}
+{% endmacro %}
+
+{%- macro bigquery__get_default_partition_by() %}
+    {% if not elementary.get_config_var("bigquery_disable_partitioning") %}
+        {% do return(
+            {
+                "field": "created_at",
+                "data_type": "timestamp",
+                "granularity": "day",
+            }
+        ) %}
+    {% endif %}
+    {% do return(none) %}
+{% endmacro %}
+
+{% macro default__get_default_partition_by() %} {% do return(none) %} {% endmacro %}

models/edr/dbt_artifacts/dbt_invocations.sql

@@ -4,6 +4,7 @@
         transient=False,
         unique_key="invocation_id",
         on_schema_change="append_new_columns",
+        partition_by=elementary.get_default_partition_by(),
         full_refresh=elementary.get_config_var("elementary_full_refresh"),
         meta={
             "timestamp_column": "created_at",

models/edr/dbt_artifacts/dbt_run_results.sql

@@ -13,6 +13,7 @@
             if target.type == "postgres"
             else []
         ),
+        partition_by=elementary.get_default_partition_by(),
         full_refresh=elementary.get_config_var("elementary_full_refresh"),
         meta={
             "dedup_by_column": "model_execution_id",