Merge pull request #16 from fivetran/connector-type-id-fix

Connector type id fix
fivetran · Apr 9, 2021 · a40369c · a40369c
2 parents 96a7754 + 7169619
commit a40369c
Show file tree

Hide file tree

Showing 5 changed files with 40 additions and 20 deletions.
diff --git a/dbt_project.yml b/dbt_project.yml
@@ -1,7 +1,7 @@
 config-version: 2
 
 name: 'fivetran_log'
-version: '0.3.0'
+version: '0.3.1'
 
 require-dbt-version: [">=0.18.0", "<0.20.0"]
 
@@ -11,6 +11,8 @@ models:
         +schema: fivetran_log
         staging:
             +schema: stg_fivetran_log
+            tmp:
+                +materialized: view
 
 
 vars:

diff --git a/integration_tests/dbt_project.yml b/integration_tests/dbt_project.yml
@@ -1,5 +1,5 @@
 name: 'fivetran_log_integration_tests'
-version: '0.3.0'
+version: '0.3.1'
 config-version: 2
 profile: 'integration_tests'
 

diff --git a/macros/get_brand_columns.sql b/macros/get_brand_columns.sql
@@ -0,0 +1,19 @@
+{% macro get_connector_columns() %}
+
+{% set columns = [
+    {"name": "_fivetran_deleted", "datatype": "boolean"},
+    {"name": "_fivetran_synced", "datatype": dbt_utils.type_timestamp()},
+    {"name": "connecting_user_id", "datatype": dbt_utils.type_string()},
+    {"name": "connector_id", "datatype": dbt_utils.type_string()},
+    {"name": "connector_name", "datatype": dbt_utils.type_string()},
+    {"name": "connector_type", "datatype": dbt_utils.type_string()},
+    {"name": "connector_type_id", "datatype": dbt_utils.type_string()},
+    {"name": "destination_id", "datatype": dbt_utils.type_string()},
+    {"name": "paused", "datatype": "boolean"},
+    {"name": "service_version", "datatype": dbt_utils.type_int()},
+    {"name": "signed_up", "datatype": dbt_utils.type_timestamp()}
+] %}
+
+{{ return(columns) }}
+
+{% endmacro %}
diff --git a/models/staging/stg_fivetran_log__connector.sql b/models/staging/stg_fivetran_log__connector.sql
@@ -1,23 +1,19 @@
 with connector as (
 
-    select * from {{ var('connector') }}
+    select * 
+    from {{ ref('stg_fivetran_log__connector_tmp') }}
 
 ),
 
 fields as (
-
-    select 
-        connector_id,
-        connector_name,
-        connector_type,  -- use coalesce(connector_type, service) if the table has a service column (deprecated)
-        destination_id,
-        connecting_user_id,
-        paused as is_paused,
-        signed_up as set_up_at,
-
-        -- Consolidating duplicate connectors (ie deleted and then re-added)
-        row_number() over ( partition by connector_name, destination_id order by _fivetran_synced desc ) as nth_last_record
-
+    select
+        {{
+            fivetran_utils.fill_staging_columns(
+                source_columns=adapter.get_columns_in_relation(ref('stg_fivetran_log__connector_tmp')),
+                staging_columns=get_connector_columns()
+            )
+        }}
+        ,row_number() over ( partition by connector_name, destination_id order by _fivetran_synced desc ) as nth_last_record
     from connector
 
 ),
@@ -27,16 +23,17 @@ final as (
     select 
         connector_id,
         connector_name,
-        connector_type, 
+        coalesce(connector_type_id, connector_type) as connector_type,
         destination_id,
         connecting_user_id,
-        is_paused,
-        set_up_at
+        paused as is_paused,
+        signed_up as set_up_at
 
     from fields
 
     -- Only look at the most recent one
     where nth_last_record = 1
 )
 
-select * from final
+select * 
+from final
diff --git a/models/staging/tmp/stg_fivetran_log__connector_tmp.sql b/models/staging/tmp/stg_fivetran_log__connector_tmp.sql
@@ -0,0 +1,2 @@
+select *
+from {{ var('connector') }}