Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

correct the condition for incremental records pick #18

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

correct the condition for incremental records pick #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4a5431f
correct the condition for incremental records pick
Jan 30, 2022
82d3921
replace - with _ in column name
Feb 2, 2022
5c54940
add camel_to_snake case conversion and column availability filter check
Feb 3, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions macros/id_map.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,27 +54,33 @@
schema = sourceName,
identifier = table_name)
-%}
{% if check_relation != None %}
{%- set elt_table_col = adapter.get_columns_in_table(schema= sourceName, identifier=table_name) -%}
{%- set elt_table_col_name = [] -%}
{%- for elt_col in elt_table_col %}
{%- set elt_table_col_name = elt_table_col_name.append(elt_col.name) %}
{%- endfor -%}
{% if check_relation != None and blotout_utils.camel_to_snake(map_column[i]) in elt_table_col_name
and blotout_utils.camel_to_snake(map_primary_key[0]) in elt_table_col_name %}
UNION
SELECT
trim("{{ blotout_utils.camel_to_snake(map_column[i]) }}") AS user_id,
trim("{{ map_primary_key[0] }}") as data_map_id,
trim("{{ blotout_utils.camel_to_snake(map_primary_key[0]) }}") as data_map_id,
'{{ map_provider[i] }}' AS "user_provider",
'{{ map_primary_provider[0] }}' AS data_map_provider,
CAST(min(etl_run_datetime) AS timestamp) AS "user_id_created"
FROM
{{ sourceName }}.{{ table_name }}
WHERE
"{{ blotout_utils.camel_to_snake(map_column[i]) }}" IS NOT NULL
AND "{{ map_primary_key[0] }}" IS NOT NULL
AND "{{ blotout_utils.camel_to_snake(map_primary_key[0]) }}" IS NOT NULL
AND trim("{{ blotout_utils.camel_to_snake(map_column[i]) }}") NOT IN ('nan', '')
AND trim("{{ map_primary_key[0] }}") NOT IN ('nan', '')
AND trim("{{ blotout_utils.camel_to_snake(map_primary_key[0]) }}") NOT IN ('nan', '')
{%- if is_incremental %}
AND CAST(etl_run_datetime AS timestamp) >
(SELECT COALESCE(MAX(user_id_created), cast('1970-01-01 00:00:00.000' as timestamp))
FROM {{ this }} WHERE orig_user_provider = '{{ map_provider[i] }}')
{% endif %}
GROUP BY "{{ map_primary_key[0] }}", "{{ blotout_utils.camel_to_snake(map_column[i]) }}"
GROUP BY "{{ blotout_utils.camel_to_snake(map_primary_key[0]) }}", "{{ blotout_utils.camel_to_snake(map_column[i]) }}"
{% endif -%}
{% endfor -%}
{% endif -%}
Expand Down Expand Up @@ -156,7 +162,7 @@
search_twclid IS NOT NULL
AND user_id IS NOT NULL
{% if is_incremental %}
AND CAST(event_datetime AS timestamp) > (select max(user_id_created) FROM {{ this }})
AND CAST(user_id_created AS timestamp) > (select max(user_id_created) FROM {{ this }})
{% endif %}
GROUP BY user_id, search_twclid
{% endif -%}
Expand Down Expand Up @@ -193,7 +199,7 @@
user_id IS NOT NULL
AND data_map_id IS NOT NULL
{%- if is_incremental %}
AND CAST(event_datetime AS TIMESTAMP) > (SELECT MAX(user_id_created) FROM {{ this }})
AND CAST(user_id_created AS TIMESTAMP) > (SELECT MAX(user_id_created) FROM {{ this }})
{% endif -%}
) emap
WHERE
Expand Down
2 changes: 1 addition & 1 deletion macros/utils.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@
{%- endif %}
{%- endfor -%}

{{ return("".join(output)) }}
{{ return("".join(output) | replace('-', '_') ) }}
{%- endmacro %}

{% macro get_records_count_in_last_interval(database_name, schema_name, table_name, interval_in_hours) %}
Expand Down