diff --git a/macros/upload_individual_datasets/upload_exposures.sql b/macros/upload_individual_datasets/upload_exposures.sql
index 6c12237f..b26cd9b1 100644
--- a/macros/upload_individual_datasets/upload_exposures.sql
+++ b/macros/upload_individual_datasets/upload_exposures.sql
@@ -189,3 +189,53 @@
         {{ return("") }}
     {% endif %}
 {%- endmacro %}
+
+
+{% macro spark__get_exposures_dml_sql(exposures) -%}
+
+    {% if exposures != [] %}
+        {% set exposure_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(6)) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(12)) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(13)) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(14)) }}
+        from values
+        {% for exposure in exposures -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ exposure.unique_id | replace("'","\\'") }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+                '{{ exposure.name | replace("'","\\'") }}', {# name #}
+                '{{ exposure.type }}', {# type #}
+                '{{ tojson(exposure.owner) }}', {# owner #}
+                '{{ exposure.maturity }}', {# maturity #}
+                '{{ exposure.original_file_path | replace('\\', '\\\\') }}', {# path #}
+                '{{ exposure.description | replace("'","\\'") }}', {# description #}
+                '{{ exposure.url }}', {# url #}
+                '{{ exposure.package_name }}', {# package_name #}
+                '{{ tojson(exposure.depends_on.nodes) }}', {# depends_on_nodes #}
+                '{{ tojson(exposure.tags) }}', {# tags #}
+                {% if var('dbt_artifacts_exclude_all_results', false) %}
+                    null
+                {% else %}
+                    '{{ tojson(exposure) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# all_results #}
+                {% endif %}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ exposure_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
\ No newline at end of file
diff --git a/macros/upload_individual_datasets/upload_invocations.sql b/macros/upload_individual_datasets/upload_invocations.sql
index 1991aeb1..6fe2c9f4 100644
--- a/macros/upload_individual_datasets/upload_invocations.sql
+++ b/macros/upload_individual_datasets/upload_invocations.sql
@@ -373,3 +373,78 @@
 
 {% endmacro -%}
 
+
+{% macro spark__get_invocations_dml_sql() -%}
+    {% set invocation_values %}
+    select
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+        {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+        nullif({{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }}, ''),
+        nullif({{ adapter.dispatch('column_identifier', 'dbt_artifacts')(12) }}, ''),
+        nullif({{ adapter.dispatch('column_identifier', 'dbt_artifacts')(13) }}, ''),
+        nullif({{ adapter.dispatch('column_identifier', 'dbt_artifacts')(14) }}, ''),
+        nullif({{ adapter.dispatch('column_identifier', 'dbt_artifacts')(15) }}, ''),
+        {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(16)) }},
+        {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(17)) }},
+        {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(18)) }},
+        {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(19)) }}
+    from values
+    (
+        '{{ invocation_id }}', {# command_invocation_id #}
+        '{{ dbt_version }}', {# dbt_version #}
+        '{{ project_name }}', {# project_name #}
+        timestamp('{{ run_started_at }}'), {# run_started_at #}
+        '{{ flags.WHICH }}', {# dbt_command #}
+        boolean('{{ flags.FULL_REFRESH }}'), {# full_refresh_flag #}
+        '{{ target.profile_name }}', {# target_profile_name #}
+        '{{ target.name }}', {# target_name #}
+        '{{ target.schema }}', {# target_schema #}
+        {{ target.threads }}, {# target_threads #}
+
+        '{{ env_var('DBT_CLOUD_PROJECT_ID', '') }}', {# dbt_cloud_project_id #}
+        '{{ env_var('DBT_CLOUD_JOB_ID', '') }}', {# dbt_cloud_job_id #}
+        '{{ env_var('DBT_CLOUD_RUN_ID', '') }}', {# dbt_cloud_run_id #}
+        '{{ env_var('DBT_CLOUD_RUN_REASON_CATEGORY', '') }}', {# dbt_cloud_run_reason_category #}
+        '{{ env_var('DBT_CLOUD_RUN_REASON', '') | replace("'","\\'") }}', {# dbt_cloud_run_reason #}
+
+        {% if var('env_vars', none) %}
+            {% set env_vars_dict = {} %}
+            {% for env_variable in var('env_vars') %}
+                {% do env_vars_dict.update({env_variable: (env_var(env_variable, '') | replace("'", "''"))}) %}
+            {% endfor %}
+            '{{ tojson(env_vars_dict) }}', {# env_vars #}
+        {% else %}
+            null, {# env_vars #}
+        {% endif %}
+
+        {% if var('dbt_vars', none) %}
+            {% set dbt_vars_dict = {} %}
+            {% for dbt_var in var('dbt_vars') %}
+                {% do dbt_vars_dict.update({dbt_var: (var(dbt_var, '') | replace("'", "''"))}) %}
+            {% endfor %}
+            '{{ tojson(dbt_vars_dict) }}', {# dbt_vars #}
+        {% else %}
+            null, {# dbt_vars #}
+        {% endif %}
+
+        '{{ tojson(invocation_args_dict) | replace('\\', '\\\\') | replace("'", "\\'") }}', {# invocation_args #}
+
+        {% set metadata_env = {} %}
+        {% for key, value in dbt_metadata_envs.items() %}
+            {% do metadata_env.update({key: (value | replace("'", "''"))}) %}
+        {% endfor %}
+        '{{ tojson(metadata_env) | replace('\\', '\\\\') }}' {# dbt_custom_envs #}
+
+    )
+    {% endset %}
+    {{ invocation_values }}
+
+{% endmacro -%}
diff --git a/macros/upload_individual_datasets/upload_model_executions.sql b/macros/upload_individual_datasets/upload_model_executions.sql
index 231f036e..6ee01a30 100644
--- a/macros/upload_individual_datasets/upload_model_executions.sql
+++ b/macros/upload_individual_datasets/upload_model_executions.sql
@@ -295,3 +295,63 @@
         {{ return("") }}
     {% endif %}
 {%- endmacro %}
+
+
+{% macro spark__get_model_executions_dml_sql(models) -%}
+    {% if models != [] %}
+        {% set model_execution_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(12) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(13) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(14) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(15) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(16)) }}
+
+        from values
+        {% for model in models -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ model.node.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+
+                {% set config_full_refresh = model.node.config.full_refresh %}
+                {% if config_full_refresh is none %}
+                    {% set config_full_refresh = flags.FULL_REFRESH %}
+                {% endif %}
+                boolean('{{ config_full_refresh }}'), {# was_full_refresh #}
+
+                '{{ model.thread_id }}', {# thread_id #}
+                '{{ model.status }}', {# status #}
+
+                {% set compile_started_at = (model.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}timestamp('{{ compile_started_at }}'){% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (model.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}timestamp('{{ query_completed_at }}'){% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ model.execution_time }}, {# total_node_runtime #}
+                null, -- rows_affected not available {# Only available in Snowflake & BigQuery #}
+                '{{ model.node.config.materialized }}', {# materialization #}
+                '{{ model.node.schema }}', {# schema #}
+                '{{ model.node.name }}', {# name #}
+                '{{ model.node.alias }}', {# alias #}
+                '{{ model.message | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}', {# message #}
+                '{{ tojson(model.adapter_response) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ model_execution_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{%- endmacro %}
diff --git a/macros/upload_individual_datasets/upload_models.sql b/macros/upload_individual_datasets/upload_models.sql
index 6e9466d3..d409b3d4 100644
--- a/macros/upload_individual_datasets/upload_models.sql
+++ b/macros/upload_individual_datasets/upload_models.sql
@@ -199,3 +199,56 @@
         {{ return("") }}
     {% endif %}
 {%- endmacro %}
+
+
+{% macro spark__get_models_dml_sql(models) -%}
+
+    {% if models != [] %}
+        {% set model_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(7)) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(12)) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(13)) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(14) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(15)) }}
+        from values
+        {% for model in models -%}
+                {% set model_copy = dbt_artifacts.copy_model(model) -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ model_copy.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+                '{{ model_copy.database }}', {# database #}
+                '{{ model_copy.schema }}', {# schema #}
+                '{{ model_copy.name }}', {# name #}
+                '{{ tojson(model_copy.depends_on.nodes) | replace('\\', '\\\\') }}', {# depends_on_nodes #}
+                '{{ model_copy.package_name }}', {# package_name #}
+                '{{ model_copy.original_file_path | replace('\\', '\\\\') }}', {# path #}
+                '{{ model_copy.checksum.checksum  | replace('\\', '\\\\') }}', {# checksum #}
+                '{{ model_copy.config.materialized }}', {# materialization #}
+                '{{ tojson(model_copy.tags) }}', {# tags #}
+                '{{ tojson(model_copy.config.meta) | replace("\\", "\\\\") | replace("'","\\'") | replace('"', '\\"') }}', {# meta #}
+                '{{ model_copy.alias }}', {# alias #}
+                {% if var('dbt_artifacts_exclude_all_results', false) %}
+                    null
+                {% else %}
+                    '{{ tojson(model_copy) | replace("\\", "\\\\") | replace("'","\\'") | replace('"', '\\"') }}' {# all_results #}
+                {% endif %}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ model_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
diff --git a/macros/upload_individual_datasets/upload_seed_executions.sql b/macros/upload_individual_datasets/upload_seed_executions.sql
index 7e8ce441..06a7df17 100644
--- a/macros/upload_individual_datasets/upload_seed_executions.sql
+++ b/macros/upload_individual_datasets/upload_seed_executions.sql
@@ -308,3 +308,61 @@
         {{ return("") }}
     {% endif %}
 {% endmacro -%}
+
+{% macro spark__get_seed_executions_dml_sql(seeds) -%}
+    {% if seeds != [] %}
+        {% set seed_execution_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(12) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(13) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(14) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(15) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(16)) }}
+        from values
+        {% for model in seeds -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ model.node.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+
+                {% set config_full_refresh = model.node.config.full_refresh %}
+                {% if config_full_refresh is none %}
+                    {% set config_full_refresh = flags.FULL_REFRESH %}
+                {% endif %}
+                boolean('{{ config_full_refresh }}'), {# was_full_refresh #}
+
+                '{{ model.thread_id }}', {# thread_id #}
+                '{{ model.status }}', {# status #}
+
+                {% set compile_started_at = (model.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}timestamp('{{ compile_started_at }}'){% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (model.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}timestamp('{{ query_completed_at }}'){% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ model.execution_time }}, {# total_node_runtime #}
+                null, -- rows_affected not available {# Only available in Snowflake #}
+                '{{ model.node.config.materialized }}', {# materialization #}
+                '{{ model.node.schema }}', {# schema #}
+                '{{ model.node.name }}', {# name #}
+                '{{ model.node.alias }}', {# alias #}
+                '{{ model.message | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}', {# message #}
+                '{{ tojson(model.adapter_response) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ seed_execution_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
\ No newline at end of file
diff --git a/macros/upload_individual_datasets/upload_seeds.sql b/macros/upload_individual_datasets/upload_seeds.sql
index 077e0516..f2905f93 100644
--- a/macros/upload_individual_datasets/upload_seeds.sql
+++ b/macros/upload_individual_datasets/upload_seeds.sql
@@ -174,3 +174,49 @@
         {{ return("") }}
     {% endif %}
 {%- endmacro %}
+
+{% macro spark__get_seeds_dml_sql(seeds) -%}
+
+    {% if seeds != [] %}
+        {% set seed_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(10)) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(12)) }}
+        from values
+        {% for seed in seeds -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ seed.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+                '{{ seed.database }}', {# database #}
+                '{{ seed.schema }}', {# schema #}
+                '{{ seed.name }}', {# name #}
+                '{{ seed.package_name }}', {# package_name #}
+                '{{ seed.original_file_path | replace('\\', '\\\\') }}', {# path #}
+                '{{ seed.checksum.checksum | replace('\\', '\\\\') }}', {# checksum #}
+                '{{ tojson(seed.config.meta) | replace("\\", "\\\\") | replace("'","\\'") | replace('"', '\\"') }}', {# meta #}
+                '{{ seed.alias }}', {# alias #}
+                {% if var('dbt_artifacts_exclude_all_results', false) %}
+                    null
+                {% else %}
+                    '{{ tojson(seed) | replace("\\", "\\\\") | replace("'","\\'") | replace('"', '\\"') }}' {# all_results #}
+                {% endif %}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ seed_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
+
diff --git a/macros/upload_individual_datasets/upload_snapshot_executions.sql b/macros/upload_individual_datasets/upload_snapshot_executions.sql
index 110a2074..0f2f73a8 100644
--- a/macros/upload_individual_datasets/upload_snapshot_executions.sql
+++ b/macros/upload_individual_datasets/upload_snapshot_executions.sql
@@ -307,3 +307,62 @@
         {{ return("") }}
     {% endif %}
 {% endmacro -%}
+
+
+{% macro spark__get_snapshot_executions_dml_sql(snapshots) -%}
+    {% if snapshots != [] %}
+        {% set snapshot_execution_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(12) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(13) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(14) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(15) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(16)) }}
+        from values
+        {% for model in snapshots -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ model.node.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+
+                {% set config_full_refresh = model.node.config.full_refresh %}
+                {% if config_full_refresh is none %}
+                    {% set config_full_refresh = flags.FULL_REFRESH %}
+                {% endif %}
+                boolean('{{ config_full_refresh }}'), {# was_full_refresh #}
+
+                '{{ model.thread_id }}', {# thread_id #}
+                '{{ model.status }}', {# status #}
+
+                {% set compile_started_at = (model.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}timestamp('{{ compile_started_at }}'){% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (model.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}timestamp('{{ query_completed_at }}'){% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ model.execution_time }}, {# total_node_runtime #}
+                null, -- rows_affected not available {# Only available in Snowflake #}
+                '{{ model.node.config.materialized }}', {# materialization #}
+                '{{ model.node.schema }}', {# schema #}
+                '{{ model.node.name }}', {# name #}
+                '{{ model.node.alias }}', {# alias #}
+                '{{ model.message | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}', {# message #}
+                '{{ tojson(model.adapter_response) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ snapshot_execution_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
\ No newline at end of file
diff --git a/macros/upload_individual_datasets/upload_snapshots.sql b/macros/upload_individual_datasets/upload_snapshots.sql
index 2d20b44b..37e5d9e0 100644
--- a/macros/upload_individual_datasets/upload_snapshots.sql
+++ b/macros/upload_individual_datasets/upload_snapshots.sql
@@ -190,3 +190,52 @@
         {{ return("") }}
     {% endif %}
 {%- endmacro %}
+
+{% macro spark__get_snapshots_dml_sql(snapshots) -%}
+
+    {% if snapshots != [] %}
+        {% set snapshot_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(7)) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(12)) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(13) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(14)) }}
+        from values
+        {% for snapshot in snapshots -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ snapshot.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+                '{{ snapshot.database }}', {# database #}
+                '{{ snapshot.schema }}', {# schema #}
+                '{{ snapshot.name }}', {# name #}
+                '{{ tojson(snapshot.depends_on.nodes) }}', {# depends_on_nodes #}
+                '{{ snapshot.package_name }}', {# package_name #}
+                '{{ snapshot.original_file_path | replace('\\', '\\\\') }}', {# path #}
+                '{{ snapshot.checksum.checksum | replace('\\', '\\\\') }}', {# checksum #}
+                '{{ snapshot.config.strategy }}', {# strategy #}
+                '{{ tojson(snapshot.config.meta) | replace("\\", "\\\\") | replace("'","\\'") | replace('"', '\\"') }}', {# meta #}
+                '{{ snapshot.alias }}', {# alias #}
+                {% if var('dbt_artifacts_exclude_all_results', false) %}
+                    null
+                {% else %}
+                    '{{ tojson(snapshot) | replace("\\", "\\\\") | replace("'","\\'") | replace('"', '\\"') }}' {# all_results #}
+                {% endif %}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ snapshot_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
diff --git a/macros/upload_individual_datasets/upload_sources.sql b/macros/upload_individual_datasets/upload_sources.sql
index c856dd67..67acd155 100644
--- a/macros/upload_individual_datasets/upload_sources.sql
+++ b/macros/upload_individual_datasets/upload_sources.sql
@@ -174,3 +174,48 @@
         {{ return("") }}
     {% endif %}
 {%- endmacro %}
+
+{% macro spark__get_sources_dml_sql(sources) -%}
+
+    {% if sources != [] %}
+        {% set source_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(11)) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(12)) }}
+        from values
+        {% for source in sources -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ source.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+                '{{ source.database }}', {# database #}
+                '{{ source.schema }}', {# schema #}
+                '{{ source.source_name }}', {# source_name #}
+                '{{ source.loader }}', {# loader #}
+                '{{ source.name }}', {# name #}
+                '{{ source.identifier }}', {# identifier #}
+                '{{ source.loaded_at_field | replace("'","\\'") }}', {# loaded_at_field #}
+                '{{ tojson(source.freshness) | replace("'","\\'") }}', {# freshness #}
+                {% if var('dbt_artifacts_exclude_all_results', false) %}
+                    null
+                {% else %}
+                    '{{ tojson(source) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# all_results #}
+                {% endif %}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ source_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
\ No newline at end of file
diff --git a/macros/upload_individual_datasets/upload_test_executions.sql b/macros/upload_individual_datasets/upload_test_executions.sql
index 568c21b6..4b1c1330 100644
--- a/macros/upload_individual_datasets/upload_test_executions.sql
+++ b/macros/upload_individual_datasets/upload_test_executions.sql
@@ -289,3 +289,55 @@
         {{ return("") }}
     {% endif %}
 {% endmacro -%}
+
+{% macro spark__get_test_executions_dml_sql(tests) -%}
+    {% if tests != [] %}
+        {% set test_execution_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(5) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(8) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(9) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(10) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(11) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(12) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(13)) }}
+        from values
+        {% for test in tests -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ test.node.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+
+                {% set config_full_refresh = test.node.config.full_refresh %}
+                {% if config_full_refresh is none %}
+                    {% set config_full_refresh = flags.FULL_REFRESH %}
+                {% endif %}
+                boolean('{{ config_full_refresh }}'), {# was_full_refresh #}
+
+                '{{ test.thread_id }}', {# thread_id #}
+                '{{ test.status }}', {# status #}
+
+                {% set compile_started_at = (test.timing | selectattr("name", "eq", "compile") | first | default({}))["started_at"] %}
+                {% if compile_started_at %}timestamp('{{ compile_started_at }}'){% else %}null{% endif %}, {# compile_started_at #}
+                {% set query_completed_at = (test.timing | selectattr("name", "eq", "execute") | first | default({}))["completed_at"] %}
+                {% if query_completed_at %}timestamp('{{ query_completed_at }}'){% else %}null{% endif %}, {# query_completed_at #}
+
+                {{ test.execution_time }}, {# total_node_runtime #}
+                null, {# rows_affected not available in Databricks #}
+                {{ 'null' if test.failures is none else test.failures }}, {# failures #}
+                '{{ test.message | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}', {# message #}
+                '{{ tojson(test.adapter_response) | replace("\\", "\\\\") | replace("'", "\\'") | replace('"', '\\"') }}' {# adapter_response #}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ test_execution_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
diff --git a/macros/upload_individual_datasets/upload_tests.sql b/macros/upload_individual_datasets/upload_tests.sql
index 0c9bbca4..c8e336a3 100644
--- a/macros/upload_individual_datasets/upload_tests.sql
+++ b/macros/upload_individual_datasets/upload_tests.sql
@@ -157,3 +157,42 @@
         {{ return("") }}
     {% endif %}
 {%- endmacro %}
+
+{% macro spark__get_tests_dml_sql(tests) -%}
+
+    {% if tests != [] %}
+        {% set test_values %}
+        select
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(1) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(2) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(3) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(4) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(5)) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(6) }},
+            {{ adapter.dispatch('column_identifier', 'dbt_artifacts')(7) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(8)) }},
+            {{ adapter.dispatch('parse_json', 'dbt_artifacts')(adapter.dispatch('column_identifier', 'dbt_artifacts')(9)) }}
+        from values
+        {% for test in tests -%}
+            (
+                '{{ invocation_id }}', {# command_invocation_id #}
+                '{{ test.unique_id }}', {# node_id #}
+                timestamp('{{ run_started_at }}'), {# run_started_at #}
+                '{{ test.name }}', {# name #}
+                '{{ tojson(test.depends_on.nodes) }}', {# depends_on_nodes #}
+                '{{ test.package_name }}', {# package_name #}
+                '{{ test.original_file_path | replace('\\', '\\\\') }}', {# test_path #}
+                '{{ tojson(test.tags) }}', {# tags #}
+                {% if var('dbt_artifacts_exclude_all_results', false) %}
+                    null
+                {% else %}
+                    '{{ tojson(test) | replace("\\", "\\\\") | replace("'","\\'") | replace('"', '\\"') }}' {# all_fields #}
+                {% endif %}
+            )
+            {%- if not loop.last %},{%- endif %}
+        {%- endfor %}
+        {% endset %}
+        {{ test_values }}
+    {% else %} {{ return("") }}
+    {% endif %}
+{% endmacro -%}
\ No newline at end of file