How to Use the Coalesce Test Stage Macro

Coalesce provides built-in data validation tests that can be configured via the Testing UI and metadata. However, sometimes more custom or complex tests are needed.

By leveraging the test_stage() built-in macro and SQL templates, you can create user-defined test nodes to validate data flows.

Test Stage Limitations

The out-of-the-box testing functionality has some limitations:

Manual configuration of test logic required
Hard to reuse tests across nodes
Limited flexibility in SQL and controls

Creating Your Test

You'll need to:

Author a macro with your test logic and name it test_<stg_name>. In the examples we use My Custom Test.
Invoke the Macro within a custom node’s Run Template.
Control failure handling behavior with a true/false clause in the stage initiation.

This tests for NULLs and does not continue the test on a failure and will fail the node execution. Using false is what will cause the failure.

{{ test_stage('My Custom Test', false) }} 

SELECT 1
FROM {{this}}
WHERE Col IS NULL

If you set the conditional to true, the test will continue the node execution after the test failure.

{{ test_stage('My Custom Test', true) }} 

SELECT 1
FROM {{this}}
WHERE Col IS NULL

Jinja Run Template Example

Full Jinja Run Template configuration for a copy of a Stage Node with two sample custom tests.

{% for test in node.tests if config.testsEnabled %}
       {% if test.runOrder == 'Before' %}
           {{ test_stage(test.name, test.continueOnFailure) }}
           {{ test.templateString }}
       {% endif %}
   {% endfor %}


{% if node.materializationType == 'table' %}
   {% if config.preSQL %}
       {{ stage('Pre-SQL') }}
       {{ config.preSQL }}
   {% endif %}
  
  
  
       {% if config.truncateBefore %}
  
           {{ stage('Truncate Stage Table') }}
           TRUNCATE IF EXISTS {{ ref_no_link(node.location.name, node.name) }}
  
       {% endif %}
  
  
       {% if config.insertStrategy in ['UNION', 'UNION ALL'] %}
           {{ stage( config.insertStrategy + ' Sources' | string ) }}
           INSERT INTO {{ ref_no_link(node.location.name, node.name) }}
               (
                   {% for col in columns %}
                       "{{ col.name }}"
                       {%- if not loop.last -%},{% endif %}
                   {% endfor %}
               )
       {% endif %}
  
  
       {% for source in sources %}
  
           {% if config.insertStrategy == 'INSERT' %}
               {{ stage('Insert ' + source.name | string ) }}
  
               INSERT INTO {{ ref_no_link(node.location.name, node.name) }}
               (
                   {% for col in source.columns %}
                       "{{ col.name }}"
                       {%- if not loop.last -%},{% endif %}
                   {% endfor %}
               )
           {% endif %}
  
           SELECT
           {% for col in source.columns %}
               {{ get_source_transform(col) }} AS "{{ col.name }}"
               {%- if not loop.last -%}, {% endif %}
           {% endfor %}
  
           {{ source.join }}
  
           {% if config.insertStrategy in ['UNION', 'UNION ALL'] and not loop.last %}
               {{config.insertStrategy}}
           {% endif %}
  
       {% endfor %}
  
   {% if config.postSQL %}
       {{ stage('Post-SQL') }}
       {{ config.postSQL }}
   {% endif %}
{% endif %}


{% if config.testsEnabled %}
   {% for test in node.tests %}
       {% if test.runOrder == 'After' %}
           {{ test_stage(test.name, test.continueOnFailure) }}
           {{ test.templateString }}
       {% endif %}
   {% endfor %}


   {% for column in columns %}
       {% for test in column.tests %}
           {{ test_stage(column.name + ": " + test.name) }}
           {{ test.templateString }}
       {% endfor %}
   {% endfor %}
  
   {{ test_stage('My Custom Test', false) }}


   SELECT 1
   FROM {{ this }}
   WHERE N_COMMENT IS NOT NULL


   {{ test_stage('My Custom Test 2', true) }}


   SELECT 1
   FROM {{ this }}
   WHERE N_REGIONKEY IS NULL


{% endif %}