https://linen.dev logo
Join Slack
Powered by
Hello Just started using Meltano and it seems like...
# plugins-dbt
o
Hello Just started using Meltano and it seems like a great tool. One trivial thing that I can’t figure out is how to do some initial index and constraints creation in the target. Im loading data from oracle into Postgres. I just want to run some sql script right after the target schema was created and before any data is pushed to it. Thank you and sorry if it’s too simple
v
Technically we could have 
target-postgres
do this, but generally we view that as something that should happen after the dataload so we'd recommend using something like 
dbt
to run whatever modifications you want to on the data after
➕ 1
some interfaces may have changed but this is my 
create_table
macro for mssql for a project that's been running for ~5 years. This is more complicated than you're asking but I think it might show why we don't mess around with this stuff in the target. Note that I also don't do any of the indexes in the raw table I do everything downstream
Copy code
{% macro sqlserver__create_table_as(temporary, relation, sql) -%}
   {%- set as_columnstore = config.get('as_columnstore', default=false) -%}
   {%- set as_temporal= config.get('as_temporal', default=false) -%}
   {% set unique_key = config.get('unique_key') %}
   {% set tmp_relation = relation.incorporate(
           path={"identifier": relation.identifier.replace("#", "") ~ '_temp_view'},
           type='view')-%}
   {%- set temp_view_sql = sql.replace("'", "''") -%}

   {{ sqlserver__drop_relation_script(tmp_relation) }}

   {{ sqlserver__drop_relation_script(relation) }}

   USE [{{ relation.database }}];
    EXEC('create view {{ tmp_relation.include(database=False) }} as
    {{ temp_view_sql }}
    ');

    SELECT * INTO {{ relation }} FROM
      {{ tmp_relation }}

   {{ sqlserver__drop_relation_script(tmp_relation) }}
    
   {% if not temporary and as_columnstore -%}
   {{ sqlserver__create_clustered_columnstore_index(relation) }}
   {% endif %}

   {% if unique_key and not as_columnstore -%}
   CREATE CLUSTERED INDEX {{ unique_key }} ON {{ relation.schema }}.{{ relation.identifier }} ({{ unique_key }})
   ALTER TABLE {{ relation.schema }}.{{ relation.identifier }} add primary key({{  unique_key  }})
   {% endif %}

   {% if as_temporal and not as_columnstore and not temporary -%}
   ALTER TABLE {{ relation.schema }}.{{ relation.identifier }} 
         ADD
          RowUpdateStatus varchar( 64 ) default 'Insert' not null
         ,UpdateUser varchar( 64 ) default system_user not null
         ,UpdateDateTime datetime2( 7 ) default sysdatetime() not null
         ,SystemStartDateTime DATETIME2 GENERATED ALWAYS AS ROW START 
            CONSTRAINT DF_SysStart_{{ relation.identifier }} DEFAULT SYSUTCDATETIME()
         ,SystemEndDateTime DATETIME2 GENERATED ALWAYS AS ROW END 
            CONSTRAINT DF_SysEnd_{{ relation.identifier }}  DEFAULT CONVERT(DATETIME2, '9999-12-31 23:59:59.9999999'),
         PERIOD FOR SYSTEM_TIME (SystemStartDateTime, SystemEndDateTime);
         
         ALTER TABLE {{ relation.schema }}.{{ relation.identifier }} 
         SET (SYSTEM_VERSIONING = ON (HISTORY_TABLE = {{ relation.schema }}.{{ relation.identifier }}_history));
   {% endif %}


{% endmacro %}_
Might help you have an
o
Thank you for the reply Let me try to understand the macros here. The reason that constraints and indexes should be added after the data load is because they exist in the source which will guarantee integrity on the target?
v
My point of sharing that macro isn't for you to have to understand my setup, just that you can set whatever you want in your target DB with DBT, just run the SQL you want so 
meltano run tap-oracle target-postgres dbt:something or dbt:run
o
Got it Thanks
np 1
m
Keep in mind there are no guarantees that the data is read/loaded in a way that guarantees constraints are maintained. If you have table A and table B where B has a foreign key to A, your pipeline could read table A, then before it reads table B a transaction could add new rows to both tables, then on reading table B the pipeline could read rows that have new references to A even though the pipeline didn't read the new rows from A previously. The next run could "catch up" when it reads A again, but for large, active databases you may see these small race conditions
Log-based replication would avoid those types of situations though, but not standard full-table or incremental strategies
o
Thank you for the clarification, this makes sense
21 Views
Open in Slack
PreviousNext
Powered by