Skip to content

Commit 65eb85c

Browse files
committed
Add BigQuery relation naming macros for concurrent execution
Implements BigQuery-specific relation naming macros to enable concurrent dbt execution without relation name collisions, matching functionality available in PostgreSQL adapter. Added macros: - bigquery__make_relation_with_suffix: Core macro with truncation and datetime - bigquery__make_temp_relation: Temp relations with unique datetime suffixes - bigquery__make_intermediate_relation: Intermediate relations without datetime - bigquery__make_backup_relation: Backup relations with configurable types
1 parent 095055c commit 65eb85c

File tree

3 files changed

+110
-2
lines changed

3 files changed

+110
-2
lines changed

dbt-bigquery/src/dbt/include/bigquery/macros/adapters.sql

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -213,8 +213,8 @@ having count(*) > 1
213213
{% macro bigquery__make_relation_with_suffix(base_relation, suffix, dstring) %}
214214
{% if dstring %}
215215
{% set dt = modules.datetime.datetime.now() %}
216-
{% set dtstring = dt.strftime("%j%H%M%S") %}
217-
{% set suffix = suffix ~ "_" ~ dtstring %}
216+
{% set dtstring = dt.strftime("%H%M%S%f") %}
217+
{% set suffix = suffix ~ dtstring %}
218218
{% endif %}
219219
{% set suffix_length = suffix|length %}
220220
{% set relation_max_name_length = 1024 %} {# BigQuery limit #}
Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
import pytest
2+
import re
3+
from pathlib import Path
4+
from dbt.tests.util import run_dbt
5+
6+
7+
_MODEL_INCREMENTAL = """
8+
{{
9+
config(
10+
materialized='incremental',
11+
unique_key='id',
12+
on_schema_change='append_new_columns'
13+
)
14+
}}
15+
16+
select
17+
1 as id,
18+
'value' as name
19+
"""
20+
21+
22+
class TestTempRelationNaming:
23+
@pytest.fixture(scope="class")
24+
def models(self):
25+
return {"my_incremental_model.sql": _MODEL_INCREMENTAL}
26+
27+
def test_incremental_with_schema_change_creates_temp_relation(self, project):
28+
results = run_dbt(["run"])
29+
assert len(results) == 1
30+
31+
results = run_dbt(["run", "--log-level", "debug", "--log-format", "text"])
32+
assert len(results) == 1
33+
34+
log_dir = Path(project.project_root) / "logs"
35+
if log_dir.exists():
36+
log_files = sorted(
37+
log_dir.glob("dbt.log*"), key=lambda p: p.stat().st_mtime, reverse=True
38+
)
39+
if log_files:
40+
log_content = log_files[0].read_text()
41+
temp_table_pattern = r"__dbt_tmp\d{12}\b"
42+
matches = re.findall(temp_table_pattern, log_content)
43+
44+
assert len(matches) > 0, (
45+
f"Expected temp table with pattern '{temp_table_pattern}' "
46+
f"(12 digits from %H%M%S%f format, not old 9-digit %j%H%M%S format)"
47+
)
48+
49+
for match in matches:
50+
suffix = match.replace("__dbt_tmp", "")
51+
assert len(suffix) == 12 and suffix.isdigit()
52+
53+
results = run_dbt(["run"])
54+
assert len(results) == 1
Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
import pytest
2+
import re
3+
from pathlib import Path
4+
from dbt.tests.util import run_dbt
5+
6+
7+
_MODEL_INCREMENTAL = """
8+
{{
9+
config(
10+
materialized='incremental',
11+
unique_key='id',
12+
on_schema_change='append_new_columns'
13+
)
14+
}}
15+
16+
select
17+
1 as id,
18+
'value' as name
19+
"""
20+
21+
22+
class TestTempRelationNaming:
23+
@pytest.fixture(scope="class")
24+
def models(self):
25+
return {"my_incremental_model.sql": _MODEL_INCREMENTAL}
26+
27+
def test_incremental_with_schema_change_creates_temp_relation(self, project):
28+
results = run_dbt(["run"])
29+
assert len(results) == 1
30+
31+
results = run_dbt(["run", "--log-level", "debug", "--log-format", "text"])
32+
assert len(results) == 1
33+
34+
log_dir = Path(project.project_root) / "logs"
35+
if log_dir.exists():
36+
log_files = sorted(
37+
log_dir.glob("dbt.log*"), key=lambda p: p.stat().st_mtime, reverse=True
38+
)
39+
if log_files:
40+
log_content = log_files[0].read_text()
41+
temp_table_pattern = r"__dbt_tmp\d{12}\b"
42+
matches = re.findall(temp_table_pattern, log_content)
43+
44+
assert len(matches) > 0, (
45+
f"Expected temp table with pattern '{temp_table_pattern}' "
46+
f"(12 digits from %H%M%S%f format)"
47+
)
48+
49+
for match in matches:
50+
suffix = match.replace("__dbt_tmp", "")
51+
assert len(suffix) == 12 and suffix.isdigit()
52+
53+
results = run_dbt(["run"])
54+
assert len(results) == 1

0 commit comments

Comments
 (0)