FivetranAI: Unified RAG working branch

integration_tests/dbt_project.yml

@@ -21,6 +21,7 @@ vars:
   rag_hubspot_engagement_company_identifier: "hubspot_engagement_company"
   rag_hubspot_engagement_contact_identifier: "hubspot_engagement_contact"
   rag_hubspot_engagement_deal_identifier: "hubspot_engagement_deal"
+  rag_hubspot_engagement_deal_company: "hubspot_deal_company"
   rag_hubspot_company_identifier: "hubspot_company"
   rag_hubspot_contact_identifier: "hubspot_contact"
   rag_hubspot_owner_identifier: "hubspot_owner"

macros/staging/hubspot/get_hubspot_deal_company_columns.sql

@@ -0,0 +1,11 @@
+{% macro get_hubspot_deal_company_columns() %}
+
+{% set columns = [
+    {"name": "_fivetran_synced", "datatype": dbt.type_timestamp()},
+    {"name": "deal_id", "datatype": dbt.type_int()},
+    {"name": "company_id", "datatype": dbt.type_int()}
+] %}
+
+{{ return(columns) }}
+
+{% endmacro %}

models/intermediate/hubspot/int_rag_hubspot__company_document.sql

@@ -0,0 +1,78 @@
+WITH owners AS (
+    SELECT
+        *,
+        COALESCE(
+            owner_email,
+            'UNKNOWN'
+        ) AS safe_email,
+        COALESCE(
+            first_name,
+            ''
+        ) AS safe_first_name,
+        COALESCE(
+            last_name,
+            ''
+        ) AS safe_last_name
+    FROM
+        {{ ref('stg_rag_hubspot__owner') }}
+),
+deals AS (
+    SELECT
+        *,
+        COALESCE({{ cast('closed_date', dbt.type_string()) }}, 'not closed yet') AS safe_close_date
+    FROM
+        {{ ref('stg_rag_hubspot__deal') }}
+),
+company AS (
+    SELECT
+        *
+    FROM
+        {{ ref('stg_rag_hubspot__company') }}
+),
+deal_company AS (
+    SELECT
+        *
+    FROM
+        {{ ref('stg_rag_hubspot__deal_company') }}
+),
+deal_descriptions AS (
+    SELECT
+        DISTINCT deals.deal_id,
+        deals.source_relation,
+        {{ dbt.concat([ "'  - {'", "'deal_name: '", "deals.title", "' // '", "'deal_owner_name: '", "owners.safe_first_name", "' '", "owners.safe_last_name", "' // '", "'deal_owner_email: '", "owners.safe_email", "' // '", "'deal_closed_date: '", "deals.safe_close_date", "'}'" ]) }} AS deal_description,
+        deals.closed_date
+    FROM
+        deals
+        JOIN owners
+        ON owners.owner_id = deals.owner_id
+        AND owners.source_relation = deals.source_relation
+),
+company_with_deal_description AS (
+    SELECT
+        company.company_id AS company_id,
+        company.source_relation AS source_relation,
+        {{ dbt.listagg(
+            measure = "dd.deal_description",
+            delimiter_text = "'\\n'",
+            order_by_clause = "order by dd.closed_date"
+        ) }} AS deal_descriptions
+    FROM
+        company
+        LEFT JOIN deal_company dc
+        ON dc.company_id = company.company_id
+        AND dc.source_relation = company.source_relation
+        LEFT JOIN deal_descriptions dd
+        ON dd.deal_id = dc.deal_id
+        AND dc.source_relation = dd.source_relation
+    GROUP BY
+        1,
+        2
+)
+SELECT
+    cdd.deal_descriptions AS deals,
+    company.*
+FROM
+    company
+    JOIN company_with_deal_description cdd
+    ON cdd.company_id = company.company_id
+    AND cdd.source_relation = company.source_relation

models/staging/hubspot_staging/src_rag_hubspot.yml

@@ -269,3 +269,16 @@ sources:
             description: The type of owner.
           - name: updated_at
             description: Timestamp representing when the owner was last updated.
+
+      - name: deal_company
+        identifier: "{{ var('rag_hubspot_deal_company_identifier', 'deal_company')}}"
+        description: Each record represents a 'link' between a deal and a company.
+        config:
+          enabled: "{{ var('rag_hubspot_sales_enabled', true) and var('rag_hubspot_company_enabled', true) and var('rag_hubspot_deal_enabled', true) }}"
+        columns:
+          - name: _fivetran_synced
+            description: '{{ doc("_fivetran_synced") }}'
+          - name: deal_id
+            description: The ID of the related contact.
+          - name: company
+            description: The ID of the related company.

models/staging/hubspot_staging/stg_rag_hubspot__company.sql

@@ -1,59 +1,29 @@
-{{ config(enabled=var('rag__using_hubspot', True)) }}
-
-with base as (
-
-    {{
-        fivetran_utils.union_data(
-            table_identifier='company', 
-            database_variable='rag_hubspot_database', 
-            schema_variable='rag_hubspot_schema', 
-            default_database=target.database,
-            default_schema='rag_hubspot',
-            default_variable='hubspot_company',
-            union_schema_variable='rag_hubspot_union_schemas',
-            union_database_variable='rag_hubspot_union_databases'
-        )
-    }}
-),
-
-fields as (
-
-    select 
-        {{
-            fivetran_utils.fill_staging_columns(
-                source_columns=adapter.get_columns_in_relation(source('rag_hubspot','company')),
-                staging_columns=get_hubspot_company_columns()
-            )
-        }}
-
-        {{ fivetran_utils.source_relation(
-            union_schema_variable='rag_hubspot_union_schemas', 
-            union_database_variable='rag_hubspot_union_databases') 
-        }}
-    from base
-),
-
-final as (
-
-    select
-        company_id,
-        source_relation,
-        is_company_deleted,
-        cast(_fivetran_synced as {{ dbt.type_timestamp() }}) as _fivetran_synced,
-        company_name,
-        description,
-        created_date,
-        industry,
-        street_address,
-        street_address_2,
-        city,
-        state,
-        country,
-        company_annual_revenue 
-
-    from fields
-
-) 
-
-select *
-from final
+{{ config(enabled = var('rag__using_hubspot', True)) }}
+
+WITH FINAL AS (
+
+    SELECT
+        {{ dbt_utils.star(
+            from = ref('stg_rag_hubspot__company_fields'),
+            except = ['id', '_fivetran_synced', 'is_deleted', 'property_name', 'property_description', 'property_createdate', 'property_industry', 'property_address', 'property_address_2', 'property_city', 'property_state', 'property_country', 'property_annualrevenue' ]
+        ) }},
+        id AS company_id,
+        CAST(_fivetran_synced AS {{ dbt.type_timestamp() }}) AS _fivetran_synced,
+        is_deleted AS is_company_deleted,
+        property_name AS company_name,
+        property_description AS description,
+        property_createdate AS created_date,
+        property_industry AS industry,
+        property_address AS street_address,
+        property_address_2 AS street_address_2,
+        property_city AS city,
+        property_state AS state,
+        property_country AS country,
+        property_annualrevenue AS company_annual_revenue
+    FROM
+        {{ ref('stg_rag_hubspot__company_fields') }}
+)
+SELECT
+    *
+FROM
+    FINAL

models/staging/hubspot_staging/stg_rag_hubspot__company_fields.sql

@@ -0,0 +1,30 @@
+{{ config(enabled=var('rag__using_hubspot', True)) }}
+
+with base as (
+
+    {{
+        fivetran_utils.union_data(
+            table_identifier='company', 
+            database_variable='rag_hubspot_database', 
+            schema_variable='rag_hubspot_schema', 
+            default_database=target.database,
+            default_schema='rag_hubspot',
+            default_variable='hubspot_company',
+            union_schema_variable='rag_hubspot_union_schemas',
+            union_database_variable='rag_hubspot_union_databases'
+        )
+    }}
+),
+
+fields as (
+
+    select 
+        *
+        {{ fivetran_utils.source_relation(
+            union_schema_variable='rag_hubspot_union_schemas', 
+            union_database_variable='rag_hubspot_union_databases') 
+        }}
+    from base
+)
+
+select * from fields

models/staging/hubspot_staging/stg_rag_hubspot__deal_company.sql

@@ -0,0 +1,46 @@
+{{ config(enabled=var('rag__using_hubspot', True)) }}
+
+with base as (
+
+    {{
+        fivetran_utils.union_data(
+            table_identifier='deal_company', 
+            database_variable='rag_hubspot_database', 
+            schema_variable='rag_hubspot_schema', 
+            default_database=target.database,
+            default_schema='rag_hubspot',
+            default_variable='hubspot_deal_company',
+            union_schema_variable='rag_hubspot_union_schemas',
+            union_database_variable='rag_hubspot_union_databases'
+        )
+    }}
+),
+
+fields as (
+
+    select 
+        {{
+            fivetran_utils.fill_staging_columns(
+                source_columns=adapter.get_columns_in_relation(source('rag_hubspot','deal_company')),
+                staging_columns=get_hubspot_deal_company_columns()
+            )
+        }}
+
+        {{ fivetran_utils.source_relation(
+            union_schema_variable='rag_hubspot_union_schemas', 
+            union_database_variable='rag_hubspot_union_databases') 
+        }}
+    from base
+),
+
+final as (
+
+    select
+        deal_id,
+        company_id,
+        source_relation
+    from fields  
+)  
+
+select *
+from final

models/staging/hubspot_staging/stg_rag_hubspot__owner.sql

@@ -21,7 +21,7 @@ fields as (
     select 
         {{
             fivetran_utils.fill_staging_columns(
-                source_columns=adapter.get_columns_in_relation(source('rag_hubspot','contact')),
+                source_columns=adapter.get_columns_in_relation(source('rag_hubspot','owner')),
                 staging_columns=get_hubspot_owner_columns()
             )
         }}