Skip to content

Commit 4f8fa6a

Browse files
MiConnellMiConnell
authored
Merge pull request #60 from flexanalytics/fix/column-upload-limit
Fix column upload limit logic
2 parents 5e558ab + f3f4eef commit 4f8fa6a

3 files changed

Lines changed: 108 additions & 145 deletions

File tree

‎macros/get_columns_content.sql‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,10 @@
11
{% macro get_columns_content(graph, resource_type) %}
22
{% set columns = [] %}
3+
{% set packages = var('dbt_observability:projects', [project_name]) %}
34

45
{% if resource_type == 'model' %}
5-
{% for model in graph.nodes.values() %}
6-
{% if model.resource_type == 'model' and model.columns %}
6+
{% for model in graph.nodes.values() | selectattr("package_name", "in", packages) %}
7+
{% if model.resource_type == 'model' and model.columns and model %}
78
{% for column in model.columns.values() %}
89
{% do columns.append({
910
'name': column.name,

‎macros/upload_columns.sql‎

Lines changed: 97 additions & 135 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
{% macro upload_columns(columns, path=None, materialization=[]) -%}
2-
{% set models = dbt_observability.get_models_list(graph, path, materialization) %}
3-
{{ return(adapter.dispatch('get_columns_dml_sql', 'dbt_observability')(columns, models)) }}
2+
{{ return(adapter.dispatch('get_columns_dml_sql', 'dbt_observability')(columns)) }}
43
{%- endmacro %}
54

65
{% macro get_observability_config_from_node(node) %}
@@ -55,13 +54,11 @@
5554

5655
{% endmacro %}
5756

58-
{% macro default__get_columns_dml_sql(columns, models) -%}
57+
{% macro default__get_columns_dml_sql(columns) -%}
58+
{% set column_values %}
5959

60-
{% if models != [] %}
61-
{% set model_values %}
62-
63-
{% set adapterArr = ['databricks','spark','snowflake'] %}
64-
{% if target.type in adapterArr %}
60+
{% set adapterArr = ['databricks','spark','snowflake'] %}
61+
{% if target.type in adapterArr %}
6562

6663
select
6764
{{ adapter.dispatch('column_identifier', 'dbt_observability')(1) }},
@@ -84,153 +81,118 @@
8481
{{ adapter.dispatch('column_identifier', 'dbt_observability')(18) }}
8582
from values
8683

87-
{% endif %}
88-
89-
{% set new_list = [] %}
90-
{% for model in models %}
91-
{% if model.columns %}
92-
{% do new_list.append( model ) %}
93-
{% endif %}
94-
{% endfor %}
95-
96-
{% for model in new_list -%}
97-
98-
{% set lowerCols = {} %}
99-
{% for k, v in model.columns.items() %}
100-
{% do lowerCols.update({k.lower(): v}) %}
101-
{% endfor %}
102-
103-
{%- set relation = adapter.get_relation(database=model.database, schema=model.schema, identifier=model.name) -%}
104-
{% set table_exists=relation is not none %}
84+
{% endif %}
10585

106-
{% if table_exists %}
107-
{%- set columns = adapter.get_columns_in_relation(relation) -%}
108-
{% endif %}
86+
{% set statsCols = [] %}
10987

88+
{% for column in columns %}
89+
{% set model = column.model %}
11090
{% set statsType = dbt_observability.get_observability_value('column_stats_type', model) %}
91+
{% set statsPick = dbt_observability.get_observability_value('column_stats_pick', model) %}
92+
{% set valsMax = dbt_observability.get_observability_value('column_values_max', model) %}
93+
{% if valsMax is none %}
94+
{% set valsMax = 10 %}
95+
{% endif %}
11196

112-
{% set statsCols = [] %}
97+
{% set isMetric = dbt_observability.is_metric(column, model) %}
11398

114-
{% if target.schema == model.schema and statsType is not none and statsType != 'NONE' %}
99+
{% if statsType == 'METRIC' %}
115100

116-
{% set statsPick = dbt_observability.get_observability_value('column_stats_pick', model) %}
117-
{% set valsMax = dbt_observability.get_observability_value('column_values_max', model) %}
118-
{% if valsMax is none %}
119-
{% set valsMax = 10 %}
101+
{% if isMetric %}
102+
{% do statsCols.append( column ) %}
120103
{% endif %}
121104

122-
{% for column in columns %}
123-
124-
{% set isMetric = dbt_observability.is_metric(column, model) %}
125-
126-
{% if statsType == 'METRIC' %}
105+
{% elif statsType == 'DOC' %}
127106

128-
{% if isMetric %}
129-
{% do statsCols.append( column ) %}
130-
{% endif %}
131-
132-
{% elif statsType == 'DOC' %}
107+
{% if column.name is defined or isMetric %}
108+
{% do statsCols.append( column ) %}
109+
{% endif %}
133110

134-
{% set col = lowerCols.get(column.name.lower()) %}
135-
{% if col.name is defined or isMetric %}
136-
{% do statsCols.append( column ) %}
137-
{% endif %}
111+
{% elif statsType == 'PICK' %}
138112

139-
{% elif statsType == 'PICK' %}
113+
{% set picked = statsPick | select('equalto', column.name.lower()) | first %}
114+
{% if picked is not none %}
115+
{% do statsCols.append( column ) %}
116+
{% endif %}
140117

141-
{% set picked = statsPick | select('equalto', column.name.lower()) | first %}
142-
{% if picked is not none %}
143-
{% do statsCols.append( column ) %}
144-
{% endif %}
118+
{% endif %}
145119

146-
{% endif %}
120+
{% endfor %}
147121

122+
{% if(statsCols|length > 0) %}
123+
124+
{%- set col_query -%}
125+
select
126+
count(1) as "row_count"
127+
{% for column in statsCols %}
128+
, count(distinct {{ column.name }}) as "{{ column.name }}_distinct"
129+
, count(1) - count({{ column.name }}) as "{{ column.name }}_null"
130+
{% if column.is_number() %}
131+
, min({{ column.name }}) as "{{ column.name }}_min"
132+
, max({{ column.name }}) as "{{ column.name }}_max"
133+
, avg({{ column.name }}) as "{{ column.name }}_avg"
134+
, sum({{ column.name }}) as "{{ column.name }}_sum"
135+
, {{ dbt_observability.stddev() }}({{ column.name }}) as "{{ column.name }}_stdev"
136+
{% else %}
137+
, null as "{{ column.name }}_min"
138+
, null as "{{ column.name }}_max"
139+
, null as "{{ column.name }}_avg"
140+
, null as "{{ column.name }}_sum"
141+
, null as "{{ column.name }}_stdev"
142+
{% endif %}
148143
{% endfor %}
149-
150-
{% if(statsCols|length > 0) %}
151-
152-
{%- set col_query -%}
153-
select
154-
count(1) as "row_count"
155-
{% for column in statsCols %}
156-
, count(distinct {{ column.name }}) as "{{ column.name }}_distinct"
157-
, count(1) - count({{ column.name }}) as "{{ column.name }}_null"
158-
{% if column.is_number() %}
159-
, min({{ column.name }}) as "{{ column.name }}_min"
160-
, max({{ column.name }}) as "{{ column.name }}_max"
161-
, avg({{ column.name }}) as "{{ column.name }}_avg"
162-
, sum({{ column.name }}) as "{{ column.name }}_sum"
163-
, {{ dbt_observability.stddev() }}({{ column.name }}) as "{{ column.name }}_stdev"
164-
{% else %}
165-
, null as "{{ column.name }}_min"
166-
, null as "{{ column.name }}_max"
167-
, null as "{{ column.name }}_avg"
168-
, null as "{{ column.name }}_sum"
169-
, null as "{{ column.name }}_stdev"
170-
{% endif %}
171-
{% endfor %}
172-
{% for column in columns %}
173-
{% set isMetric = dbt_observability.is_metric(column, model) %}
174-
{% set col = lowerCols.get(column.name.lower()) %}
175-
{% if statsType!='METRIC' and column.is_string() and col.name is defined and not isMetric and target.type != 'redshift' %}
176-
, {{ dbt.listagg("distinct "~column.name, "', '", "order by "~column.name, valsMax) }} as "{{ column.name }}_values"
177-
{% else %}
178-
, null as "{{ column.name }}_values"
179-
{% endif %}
180-
{% endfor %}
181-
from {{relation}}
182-
{%- endset -%}
183-
{%- set results = run_query(col_query) -%}
184-
144+
{% for column in columns %}
145+
{% set isMetric = dbt_observability.is_metric(column, model) %}
146+
{% if statsType!='METRIC' and column.is_string() and column.name is defined and not isMetric and target.type != 'redshift' %}
147+
, {{ dbt.listagg("distinct "~column.name, "', '", "order by "~column.name, valsMax) }} as "{{ column.name }}_values"
185148
{% else %}
149+
, null as "{{ column.name }}_values"
150+
{% endif %}
151+
{% endfor %}
152+
from {{relation}}
153+
{%- endset -%}
154+
{%- set results = run_query(col_query) -%}
186155

187-
{%- set results = none -%}
156+
{% else %}
188157

189-
{% endif %}
158+
{%- set results = none -%}
190159

191-
{% else %}
160+
{% endif %}
192161

193-
{%- set results = none -%}
194162

163+
{% for column in columns %}
164+
{% set metric = graph.metrics.values() | selectattr('expression'|lower, 'equalto', column.name.lower()) | first %}
165+
{% set isMetric = dbt_observability.is_metric(column, model) %}
166+
(
167+
'{{ invocation_id }}' {# command_invocation_id #}
168+
, '{{ column.model.unique_id }}' {# node_id #}
169+
, '{{ column.name }}' {# column_name #}
170+
, '{{ column.data_type }}' {# data_type #}
171+
{% set column_meta = adapter.dispatch('escape_singlequote', 'dbt_observability')(tojson(column.meta)) %}
172+
{% if target.type == 'bigquery' %}
173+
, {{ '[]' if column.tags is not defined else tojson(column.tags) }} {# tags #}
174+
, parse_json('{{ '{}' if column.meta is not defined else column_meta }}') {# meta #}
175+
{% else %}
176+
, '{{ null if column.tags is not defined else tojson(column.tags) }}' {# tags #}
177+
, '{{ null if column.meta is not defined else column_meta }}' {# meta #}
195178
{% endif %}
196-
197-
{% for column in columns %}
198-
{% set metric = graph.metrics.values() | selectattr('expression'|lower, 'equalto', column.name.lower()) | first %}
199-
{% set isMetric = dbt_observability.is_metric(column, model) %}
200-
{% set col = lowerCols.get(column.name.lower()) %}
201-
(
202-
'{{ invocation_id }}' {# command_invocation_id #}
203-
, '{{ model.unique_id }}' {# node_id #}
204-
, '{{ column.name }}' {# column_name #}
205-
, '{{ column.data_type }}' {# data_type #}
206-
{% if target.type == 'bigquery' %}
207-
, {{ '[]' if col.tags is not defined else tojson(col.tags) }} {# tags #}
208-
, parse_json('{{ '{}' if col.meta is not defined else tojson(col.meta) }}') {# meta #}
209-
{% else %}
210-
, '{{ null if col.tags is not defined else tojson(col.tags) }}' {# tags #}
211-
, '{{ null if col.meta is not defined else tojson(col.meta) }}' {# meta #}
212-
{% endif %}
213-
, '{{ null if col.description is not defined else adapter.dispatch('escape_singlequote', 'dbt_observability')(col.description)}}' {# description #}
214-
, '{{ "Y" if col.name is defined or isMetric else "N" }}' {# is_documented #}
215-
{% set statsCol = statsCols | selectattr('name', 'equalto', column.name) | first %}
216-
, {{ (results and results.columns['row_count'].values()[0]) or 'null' }} {# row_count #}
217-
, {{ (results and results.columns[column.name ~ '_distinct'] and results.columns[column.name ~ '_distinct'].values()[0]) or 'null' }} {# row_distinct #}
218-
, {{ (results and results.columns[column.name ~ '_null'] and results.columns[column.name ~ '_null'].values()[0]) or 'null' }} {# row_null #}
219-
, {{ (results and results.columns[column.name ~ '_min'] and results.columns[column.name ~ '_min'].values()[0]) or 'null' }} {# row_min #}
220-
, {{ (results and results.columns[column.name ~ '_max'] and results.columns[column.name ~ '_max'].values()[0]) or 'null' }} {# row_max #}
221-
, {{ (results and results.columns[column.name ~ '_avg'] and results.columns[column.name ~ '_avg'].values()[0]) or 'null' }} {# row_avg #}
222-
, {{ (results and results.columns[column.name ~ '_sum'] and results.columns[column.name ~ '_sum'].values()[0]) or 'null' }} {# row_sum #}
223-
, {{ (results and results.columns[column.name ~ '_stdev'] and results.columns[column.name ~ '_stdev'].values()[0]) or 'null' }} {# row_stdev #}
224-
, '{{ "Y" if isMetric else "N" }}' {# is_metric #}
225-
, '{{ (results and results.columns[column.name ~ '_values'] and results.columns[column.name ~ '_values'].values()[0]) or null }}' {# column_values #}
226-
)
227-
{%- if not loop.last %},{%- endif %}
228-
{% endfor %}
229-
{%- if not loop.last %},{%- endif %}
230-
{%- endfor %}
231-
{% endset %}
232-
{{ model_values }}
233-
{% else %}
234-
{{ return("") }}
235-
{% endif %}
179+
{% set column_description = adapter.dispatch('escape_singlequote', 'dbt_observability')(column.description) %}
180+
, '{{ null if column.description is not defined else column_description }}' {# description #}
181+
, '{{ "Y" if column.name is defined or isMetric else "N" }}' {# is_documented #}
182+
{% set statsCol = statsCols | selectattr('name', 'equalto', column.name) | first %}
183+
, {{ (results and results.columns['row_count'].values()[0]) or 'null' }} {# row_count #}
184+
, {{ (results and results.columns[column.name ~ '_distinct'] and results.columns[column.name ~ '_distinct'].values()[0]) or 'null' }} {# row_distinct #}
185+
, {{ (results and results.columns[column.name ~ '_null'] and results.columns[column.name ~ '_null'].values()[0]) or 'null' }} {# row_null #}
186+
, {{ (results and results.columns[column.name ~ '_min'] and results.columns[column.name ~ '_min'].values()[0]) or 'null' }} {# row_min #}
187+
, {{ (results and results.columns[column.name ~ '_max'] and results.columns[column.name ~ '_max'].values()[0]) or 'null' }} {# row_max #}
188+
, {{ (results and results.columns[column.name ~ '_avg'] and results.columns[column.name ~ '_avg'].values()[0]) or 'null' }} {# row_avg #}
189+
, {{ (results and results.columns[column.name ~ '_sum'] and results.columns[column.name ~ '_sum'].values()[0]) or 'null' }} {# row_sum #}
190+
, {{ (results and results.columns[column.name ~ '_stdev'] and results.columns[column.name ~ '_stdev'].values()[0]) or 'null' }} {# row_stdev #}
191+
, '{{ "Y" if isMetric else "N" }}' {# is_metric #}
192+
, '{{ (results and results.columns[column.name ~ '_values'] and results.columns[column.name ~ '_values'].values()[0]) or null }}' {# column_values #}
193+
)
194+
{%- if not loop.last %},{%- endif %}
195+
{% endfor %}
196+
{% endset %}
197+
{{ column_values }}
236198
{% endmacro -%}

‎macros/upload_results.sql‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -64,27 +64,27 @@
6464
}}
6565

6666
{% do log("Uploading model columns", true) %}
67-
{% set models = dbt_observability.get_relation('columns') %}
67+
{% set column_table = dbt_observability.get_relation('columns') %}
6868
{% set columns = dbt_observability.get_columns_content(graph, 'model') %}
6969
{% for i in range(0, columns | length, upload_limit) %}
7070
{% set content_columns = dbt_observability.upload_columns(columns[i: i + upload_limit], path, materialization) %}
7171
{{ dbt_observability.insert_into_metadata_table(
72-
database_name=models.database,
73-
schema_name=models.schema,
74-
table_name=models.identifier,
72+
database_name=column_table.database,
73+
schema_name=column_table.schema,
74+
table_name=column_table.identifier,
7575
content=content_columns
7676
) }}
7777
{% endfor %}
7878

7979
{% do log("Uploading source columns", true) %}
80-
{% set models = dbt_observability.get_relation('columns') %}
80+
{% set source_column_table = dbt_observability.get_relation('columns') %}
8181
{% set columns = dbt_observability.get_columns_content(graph, 'source') %}
8282
{% for i in range(0, columns | length, upload_limit) %}
8383
{% set content_sources = dbt_observability.upload_source_schema(graph, columns[i: i + upload_limit]) %}
8484
{{ dbt_observability.insert_into_metadata_table(
85-
database_name=models.database,
86-
schema_name=models.schema,
87-
table_name=models.identifier,
85+
database_name=source_column_table.database,
86+
schema_name=source_column_table.schema,
87+
table_name=source_column_table.identifier,
8888
content=content_sources
8989
) }}
9090
{% endfor %}

0 commit comments

Comments
 (0)