dbt-labs / dbt-labs/dbt-external-tables
Add column.expression to snowpipe get_copy.sql
- Dominant language
- Jupyter Notebook
- Stars
- 384
- Forks
- 150
- Avg merge
- 1h 25m
- Merged PRs (30d)
- 1
Description
### Describe the feature
Add the expression to the plugins/snowflake/snowpipe/get_copy.sql code so that the external_stage_source.yml expressions can be utilised for snowpipe.
I've tested this without issue:
```sql
{% macro snowflake_get_copy_sql(source_node, explicit_transaction=false) %}
{# This assumes you have already created an external stage #}
{%- set columns = source_node.columns.values() -%}
{%- set external = source_node.external -%}
{%- set is_csv = dbt_external_tables.is_csv(external.file_format) %}
{%- set copy_options = external.snowpipe.get('copy_options', none) -%}
{%- if explicit_transaction -%} begin; {%- endif %}
copy into {{source(source_node.source_name, source_node.name)}}
from (
select
{% if columns|length == 0 %}
$1::variant as value,
{% else -%}
{%- for column in columns -%}
{%- set col_expression -%}
{%- if column.expression -%}
{{column.expression}}
{%- else -%}
{%- if is_csv -%}nullif(${{loop.index}},''){# special case: get columns by ordinal position #}
{%- else -%}nullif($1:{{column.name}},''){# standard behavior: get columns by name #}
{%- endif -%}
{%- endif -%}
{%- endset -%}
{{col_expression}}::{{column.data_type}} as {{column.name}},
{% endfor -%}
{% endif %}
metadata$filename::varchar as metadata_filename,
metadata$file_row_number::bigint as metadata_file_row_number,
metadata$file_last_modified::timestamp as metadata_file_last_modified,
metadata$start_scan_time::timestamp as _dbt_copied_at
from {{external.location}} {# stage #}
)
file_format = {{external.file_format}}
{% if external.pattern -%} pattern = '{{external.pattern}}' {%- endif %}
{% if copy_options %} {{copy_options}} {% endif %};
{% if explicit_transaction -%} commit; {%- endif -%}
{% endmacro %}
```
### Describe alternatives you've considered
n/a
### Additional context
dbt-external-tables
prod_analytics/dbt_packages/dbt_external_tables/macros/plugins/snowflake/snowpipe/get_copy_sql.sql
I took the lines of code from snowflake/create_external_table.sql that pulls the column.expressions from the yml file.
The version I've added above will create the copy into statement dynamically with the column expression included, as opposed to iterating over the column positions
e.g.
```yaml
columns:
- name: REPORT_DATE
expression: right(left($1::VARCHAR,171),8)
data_type: VARCHAR
- name: MERCHANT_NUMBER
expression: ltrim(rtrim(right(left($2::VARCHAR,48),20)))
data_type: VARCHAR
```
Generated copy command:
```sql
copy into src_prd.FIRST_DATA_SOURCE.FD_NORTH_SETTLE_003
from (
select
right(left($1::VARCHAR,171),8)::VARCHAR as REPORT_DATE,
ltrim(rtrim(right(left($2::VARCHAR,48),20)))::VARCHAR as MERCHANT_NUMBER,
metadata$filename::varchar as metadata_filename,
metadata$file_row_number::bigint as metadata_file_row_number,
metadata$file_last_modified::timestamp as metadata_file_last_modified,
metadata$start_scan_time::timestamp as _dbt_copied_at
from @ADMIN.PASSAGE_S3_STAGE
)
file_format = ADMIN.cl9dfmde_v1_ingest
pattern = '.*CL9DFMDE.36829.NAGW-YHFBR001.708980000887.d.20210416.dfm.20231206.*'
```
Copy into generated prior to command:
```sql
copy into src_prd.FIRST_DATA_SOURCE.FD_NORTH_SETTLE_003
from (
select
$1::VARCHAR as REPORT_DATE,
$2::VARCHAR as MERCHANT_NUMBER,
metadata$filename::varchar as metadata_filename,
metadata$file_row_number::bigint as metadata_file_row_number,
metadata$file_last_modified::timestamp as metadata_file_last_modified,
metadata$start_scan_time::timestamp as _dbt_copied_at
from @ADMIN.PASSAGE_S3_STAGE
)
file_format = ADMIN.cl9dfmde_v1_ingest
pattern = **'.*CL9DFMDE.36829.NAGW-YHFBR001.708980000887.d.20210416.dfm.20231206.*'**
```
### Who will this benefit?
Everyone using external stages with snowpipe
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.