|
| 1 | +{% macro _datashare_sql_string(value) %} |
| 2 | + {{ return("'" ~ (value | string | replace("'", "''")) ~ "'") }} |
| 3 | +{%- endmacro -%} |
| 4 | + |
| 5 | +{% macro _datashare_unique_key_columns_sql(unique_key_columns) %} |
| 6 | + {%- if unique_key_columns is string -%} |
| 7 | + {%- set unique_key_columns = [unique_key_columns] -%} |
| 8 | + {%- elif unique_key_columns is not iterable or unique_key_columns is mapping -%} |
| 9 | + {{ return("CAST(ARRAY[] AS ARRAY(VARCHAR))") }} |
| 10 | + {%- endif -%} |
| 11 | + {%- set quoted = [] -%} |
| 12 | + {%- for col in unique_key_columns -%} |
| 13 | + {%- do quoted.append(_datashare_sql_string(col)) -%} |
| 14 | + {%- endfor -%} |
| 15 | + {{ return("CAST(ARRAY[] AS ARRAY(VARCHAR))" if quoted | length == 0 else "ARRAY[" ~ quoted | join(', ') ~ "]") }} |
| 16 | +{%- endmacro -%} |
| 17 | + |
| 18 | +{% macro _datashare_optional_time_sql(value) %} |
| 19 | + {{ return('NULL' if value is none else 'CAST(' ~ value ~ ' AS VARCHAR)') }} |
| 20 | +{%- endmacro -%} |
| 21 | + |
| 22 | +{# |
| 23 | + Datashare sync macro - generates ALTER TABLE ... EXECUTE datashare() SQL. |
| 24 | + Config reference and usage: docs/dune-datashares.md |
| 25 | +#} |
| 26 | +{% macro _datashare_table_sync_sql( |
| 27 | + schema_name |
| 28 | + , table_name |
| 29 | + , meta |
| 30 | + , materialized |
| 31 | + , unique_key=None |
| 32 | + , time_start=None |
| 33 | + , time_end=None |
| 34 | + , full_refresh=False |
| 35 | + , catalog_name=target.database |
| 36 | +) %} |
| 37 | + {%- set model_ref = schema_name ~ '.' ~ table_name -%} |
| 38 | + {%- if meta is not mapping or meta.get('datashare') is none or meta.get('datashare') is not mapping -%} |
| 39 | + {{ log('Skipping datashare sync for ' ~ model_ref ~ ': meta.datashare is not configured.', info=True) }} |
| 40 | + {{ return(none) }} |
| 41 | + {%- endif -%} |
| 42 | + {%- set datashare = meta.get('datashare') -%} |
| 43 | + {%- if datashare.get('enabled') is not sameas true -%} |
| 44 | + {{ log('Skipping datashare sync for ' ~ model_ref ~ ': meta.datashare.enabled is not true.', info=True) }} |
| 45 | + {{ return(none) }} |
| 46 | + {%- endif -%} |
| 47 | + {%- if materialized not in ['incremental', 'table'] -%} |
| 48 | + {{ log('Skipping datashare sync for ' ~ model_ref ~ ': materialization "' ~ materialized ~ '" is not incremental/table.') }} |
| 49 | + {{ return(none) }} |
| 50 | + {%- endif -%} |
| 51 | + {%- set time_column = datashare.get('time_column') -%} |
| 52 | + {%- set resolved_time_start = time_start if time_start is not none else datashare.get('time_start') -%} |
| 53 | + {%- set resolved_time_end = time_end if time_end is not none else datashare.get('time_end', 'now()') -%} |
| 54 | + |
| 55 | + {%- set sql -%} |
| 56 | +ALTER TABLE {{ catalog_name }}.{{ schema_name }}.{{ table_name }} EXECUTE datashare( |
| 57 | + time_column => {{ _datashare_sql_string(time_column | default('', true)) }}, |
| 58 | + unique_key_columns => {{ _datashare_unique_key_columns_sql(datashare.get('unique_key_columns', unique_key)) }}, |
| 59 | + time_start => {{ _datashare_optional_time_sql(resolved_time_start) }}, |
| 60 | + time_end => {{ _datashare_optional_time_sql(resolved_time_end) }}, |
| 61 | + full_refresh => {{ 'true' if full_refresh else 'false' }} |
| 62 | +) |
| 63 | + {%- endset -%} |
| 64 | + {{ log('datashare sync preview for ' ~ model_ref ~ ':\n' ~ sql, info=True) }} |
| 65 | + {{ return(sql) }} |
| 66 | +{%- endmacro -%} |
| 67 | + |
| 68 | +{% macro datashare_trigger_sync() %} |
| 69 | + {%- if target.name != 'prod' -%} |
| 70 | + {{ log('Skipping datashare sync for ' ~ this.schema ~ '.' ~ this.identifier ~ ': datashare post-hook only runs on the prod target.', info=True) }} |
| 71 | + {{ return('') }} |
| 72 | + {%- endif -%} |
| 73 | + {{ return(_datashare_table_sync_sql( |
| 74 | + schema_name=this.schema, |
| 75 | + table_name=this.identifier, |
| 76 | + meta=model.config.get('meta', {}), |
| 77 | + materialized=model.config.materialized, |
| 78 | + unique_key=model.config.get('unique_key'), |
| 79 | + full_refresh=(not is_incremental()) |
| 80 | + ) or '') }} |
| 81 | +{%- endmacro -%} |
| 82 | + |
| 83 | +{% macro _datashare_resolve_model_node(model_selector) %} |
| 84 | + {%- set matches = [] -%} |
| 85 | + {%- for node in graph.nodes.values() -%} |
| 86 | + {%- if node.resource_type == 'model' -%} |
| 87 | + {%- set fqn_name = node.fqn | join('.') -%} |
| 88 | + {%- if node.unique_id == model_selector or node.name == model_selector or node.alias == model_selector or fqn_name == model_selector -%} |
| 89 | + {%- do matches.append(node) -%} |
| 90 | + {%- endif -%} |
| 91 | + {%- endif -%} |
| 92 | + {%- endfor -%} |
| 93 | + |
| 94 | + {%- if matches | length == 0 -%} |
| 95 | + {{ exceptions.raise_compiler_error("No model found for selector '" ~ model_selector ~ "'. Use model name, alias, fqn, or unique_id.") }} |
| 96 | + {%- endif -%} |
| 97 | + |
| 98 | + {%- if matches | length > 1 -%} |
| 99 | + {{ exceptions.raise_compiler_error("Model selector '" ~ model_selector ~ "' is ambiguous. Matches: " ~ (matches | map(attribute='unique_id') | join(', '))) }} |
| 100 | + {%- endif -%} |
| 101 | + |
| 102 | + {{ return(matches[0]) }} |
| 103 | +{%- endmacro -%} |
| 104 | + |
| 105 | +{% macro datashare_trigger_sync_operation(model_selector, time_start=None, time_end=None, dry_run=False, full_refresh=False) %} |
| 106 | + {%- set node = _datashare_resolve_model_node(model_selector) -%} |
| 107 | + {%- set node_config = node.config if node.config is mapping else {} -%} |
| 108 | + {%- set materialized = node_config.get('materialized', 'view') -%} |
| 109 | + {%- set table_name = node.alias if node.alias is not none else node.name -%} |
| 110 | + {%- set is_full_refresh = materialized == 'table' or full_refresh is sameas true -%} |
| 111 | + |
| 112 | + {%- set sql = _datashare_table_sync_sql( |
| 113 | + schema_name=node.schema, |
| 114 | + table_name=table_name, |
| 115 | + meta=node_config.get('meta', {}), |
| 116 | + materialized=materialized, |
| 117 | + unique_key=node_config.get('unique_key'), |
| 118 | + time_start=time_start, |
| 119 | + time_end=time_end, |
| 120 | + full_refresh=is_full_refresh, |
| 121 | + catalog_name=node.database or target.database |
| 122 | + ) -%} |
| 123 | + |
| 124 | + {%- if sql is none -%} |
| 125 | + {{ exceptions.raise_compiler_error("Cannot sync " ~ node.schema ~ "." ~ table_name ~ ": model must be incremental or table with meta.datashare.enabled = true.") }} |
| 126 | + {%- endif -%} |
| 127 | + |
| 128 | + {%- set is_dry_run = dry_run is sameas true or (dry_run is string and dry_run | lower in ['true', '1', 'yes', 'y']) -%} |
| 129 | + {%- if not is_dry_run -%} |
| 130 | + {% do run_query(sql) %} |
| 131 | + {{ log('Executed datashare sync for selector ' ~ model_selector, info=True) }} |
| 132 | + {%- endif -%} |
| 133 | + {{ return(sql) }} |
| 134 | +{%- endmacro -%} |
0 commit comments