create.sql.j2

{#
    Copyright (c) 2023 Coalesce. All rights reserved.
This script and its associated documentation are confidential and proprietary to Coalesce.
Unauthorized reproduction, distribution, or disclosure of this material is strictly prohibited.
Coalesce permits you to copy and modify this script for the purposes of using with Coalsce but
does not permit copying or modification for any other purpose.  
#}
{# == Node Type Version        : 1  == #}
{# == Node Type Name           : CopyInto with Full Load Option == #}
{# == Node Type Description    : This node creates target table to load data from internal or external stages == #}
{#Release date:20240531 Version:1.0.0#}
{#Release date:20240822  Version:1.1.1 Released By:AS  Description: Materialization type transient table added #}
{#Release date:20241025  Version:1.1.4 Released By:AS  Description: Unified Storage and file format inputs #}
{#Release date:20241105  Version:1.1.5 Released By:AS  Description: Truncate option added in config #}
{#Release date:20250109  Version:1.1.8 Released By:AS  Description:  Path config added under Source data and new config added for PARQUET fileformat#}



{% if desiredState %}

    
    {% if desiredState.node.materializationType == 'table' or desiredState.node.materializationType == 'transient table'  %}     

        {% if currentState != undefined and currentState.node.materializationType != desiredState.node.materializationType %}

                {# Change in materialization type #}

                {{ stage('Drop ' + currentState.node.materializationType + ' ' + ref_no_link(currentState.node.location.name, currentState.node.name)) }}

                 {% if currentState.node.materializationType == 'view' %} 
                      DROP {{currentState.node.materializationType}} IF EXISTS {{ ref_no_link(currentState.node.location.name, currentState.node.name) }}
                {% else %}
                      DROP TABLE IF EXISTS {{ ref_no_link(currentState.node.location.name, currentState.node.name) }}
                {% endif %}
                 
                {{ stage('Create'+ ' {{desiredState.node.materializationType}}') }}
               CREATE OR REPLACE {{desiredState.node.materializationType}} {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
                (
                {% for col in desiredState.columns %}
                    {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, col.name) }} {{ col.dataType }}
                    {%- if not col.nullable %} NOT NULL
                        {%- if col.defaultValue | length > 0 %} DEFAULT {{ col.defaultValue }}{% endif %}
                    {% endif %}
                    {%- if col.description | length > 0 %} COMMENT '{{ col.description | escape }}'{% endif %}
                    {%- if not loop.last -%}, {% endif %}
                {% endfor %}
                )
               {%- if desiredState.node.description | length > 0 %} COMMENT = '{{ desiredState.node.description | escape }}'{% endif %}

                
        {% endif %}

        {% if currentState == undefined or currentState.storageLocations|length == 0 %}
            					  
			{%if desiredState.config.inferschema == true %}

       
					       {% set srcSchName = desiredState.node.location.name %}
                           {% set db = ref_no_link(desiredState.node.location.name, desiredState.node.name).split('.')[0]%}
                           {% set sch = ref_no_link(desiredState.node.location.name, desiredState.node.name).split('.')[1]%}

                           {% set stageStorageLocationName = desiredState.config.stageStorageLocationName %}
                           {% set dbStage =  ref_no_link(desiredState.config.stageStorageLocationName, desiredState.node.name).split('.')[0] %}
                           {% set schStage =  ref_no_link(desiredState.config.stageStorageLocationName, desiredState.node.name).split('.')[1]%}

                            # File format locations
                         {%-if desiredState.config.fileFormatType == 'File Format Name'-%}
                           {% set fileFormatLocationName = desiredState.config.fileFormatLoc %}
                           {% set dbff = desiredState.storageLocations | selectattr('name', 'equalto',fileFormatLocationName) | map(attribute='database') | first %}
                           {% set schff = desiredState.storageLocations | selectattr('name', 'equalto',fileFormatLocationName) | map(attribute='schema') | first %}
                        {%else%}
                            {% set fileFormatLocationName = desiredState.node.location.name %}
                           {% set dbff = desiredState.storageLocations | selectattr('name', 'equalto',fileFormatLocationName) | map(attribute='database') | first %}
                           {% set schff = desiredState.storageLocations | selectattr('name', 'equalto',fileFormatLocationName) | map(attribute='schema') | first %}
                        {%endif%}
                 
                            # File format sql
                         {%-if desiredState.config.fileFormatType == 'File Format Values'-%}
                            {%set fileformatsql = "'CREATE OR REPLACE TEMPORARY FILE FORMAT " +"\""+dbff+"\".\""+schff+"\".\"TEMP_"+desiredState.config.fileType+"_"+desiredState.node.name+"\""%}

                            {%set fileformatsql = fileformatsql + "\n"+ "TYPE" + " =''"+ desiredState.config.fileType+"''"%}
                            {% if desiredState.config.compression != "" %}
                                {%set fileformatsql = fileformatsql + "\n"+ "COMPRESSION"+ "=''"+ desiredState.config.compression+"''"%}
                            {%endif%}		 
                            {% if desiredState.config.recDelim and desiredState.config.fileType in ('CSV')%}
                               {%set fileformatsql = fileformatsql + "\n"+ "RECORD_DELIMITER"+ "=''"+ desiredState.config.recDelim+"''"%}
                            {%endif%}
                            {% if desiredState.config.fieldDelim and desiredState.config.fileType in ('CSV')%}
                                {%set fileformatsql = fileformatsql + "\n"+ "FIELD_DELIMITER"+ "=''"+ desiredState.config.fieldDelim+"''"%}
                            {%endif%}

							{% if desiredState.config.fileType in ('CSV') and desiredState.config.fieldEnclosed %}
                               {% set val = desiredState.config.fieldEnclosed %}
							   {% set fileformatsql = fileformatsql + "\nFIELD_OPTIONALLY_ENCLOSED_BY=" + ( "NONE" if val | upper == "NONE" 
							   else "''\\" ~ val ~ "''" ) %}
                            {% endif %}
                            {% if desiredState.config.parsehdr and  desiredState.config.fileType in ('CSV')%}
                              {%set fileformatsql = fileformatsql + "\n"+ "PARSE_HEADER" + "="+ "TRUE"+""%}
                            {%endif%}
                            {% if (desiredState.config.encoding != "" and desiredState.config.encoding is not none)and  desiredState.config.fileType in ('CSV') %}
                                {%set fileformatsql = fileformatsql + "\n"+ "ENCODING"+ "=''"+ desiredState.config.encoding+"''"%}
                            {%endif%}
                            {% if (desiredState.config.binformat != "" and desiredState.config.binformat is not none )and  desiredState.config.fileType in ('CSV') %}
                                {%set fileformatsql = fileformatsql + "\n"+ "BINARY_FORMAT"+ "=''"+ desiredState.config.binformat+"''"%}
                            {%endif%}
                            {% if desiredState.config.replaceinvalidcharacters%}
                                 {%set fileformatsql = fileformatsql + "\n"+ "REPLACE_INVALID_CHARACTERS" + "="+ "TRUE" +""%}
                            {%endif%}
                            {%if desiredState.config.columnmismatch and desiredState.config.fileType in ('CSV')%}
                                {%set fileformatsql = fileformatsql + "\n"+ "ERROR_ON_COLUMN_COUNT_MISMATCH" + "="+ "TRUE" +""%}
                            {%endif%}
                            {% if desiredState.config.stripouterarray == true and desiredState.config.fileType in ('JSON')%}
                              {%set fileformatsql = fileformatsql + "\n"+ "STRIP_OUTER_ARRAY" + "="+ "TRUE" +""%}
                            {%endif%}
                            {%if  desiredState.config.usevectscan == true and  desiredState.config.fileType in ('PARQUET')%}
                              {%set fileformatsql = fileformatsql + "\n"+ "USE_VECTORIZED_SCANNER" + "="+ "TRUE" +""%}
                            {%endif%}
                            {% if desiredState.config.emptyfield and  desiredState.config.fileType in ('CSV')%}
                              {%set fileformatsql = fileformatsql + "\n"+ "EMPTY_FIELD_AS_NULL" + "="+ "TRUE"+""%}
                            {%endif%}
                            {% if desiredState.config.bytemark and  desiredState.config.fileType in ('CSV')%}
                              {%set fileformatsql = fileformatsql + "\n"+ "SKIP_BYTE_ORDER_MARK" + "="+ "TRUE"+""%}
                            {%endif%}
							
							{% if desiredState.config.fileType in ('CSV') and desiredState.config.escapeenclosed is not none and desiredState.config.escapeenclosed != "" %}
                               {% set val = desiredState.config.escapeenclosed %}
                               {% set fileformatsql = fileformatsql + "\nESCAPE_UNENCLOSED_FIELD=" + ( "NONE" if val | upper == "NONE" 
							   else "''\\" ~ val ~ "''" ) %}
                            {% endif %}

							
							{% if desiredState.config.fileType in ('CSV') and desiredState.config.escapeval is not none and desiredState.config.escapeval != "" %}
                               {% set val = desiredState.config.escapeval %}
                               {% set fileformatsql = fileformatsql + "\nESCAPE=" + ( "NONE" if val | upper == "NONE" 
                               else "''\\" ~ val ~ "''" ) %}
                            {% endif %}

                            {% if desiredState.config.mulline and  desiredState.config.fileType in ('CSV')%}
                              {%set fileformatsql = fileformatsql + "\n"+ "MULTI_LINE" + "="+ "TRUE"+""%}
                            {%endif%}
                            {%set fileformatsql =  fileformatsql + "\n"+ "'" %}
                          {%endif%}   
                           
                        {#Subfolder path #}
                        {%set subf = ""%}
                        {% set subfolder_value = desiredState.config.subfolder | default('') %}
                        {%if desiredState.config.subfolder == "" %}
                            {%set subf = "" %}
                        {%elif desiredState.config.subfolder == None%}
                            {%set subf = "" %}
                        {%else%}
                          {%set subf = "/"+ subfolder_value %}
                        {%endif%}

                           {%- if ","  in desiredState.config.fileNames %} 
                               {% set formatted_variable = desiredState.config.fileNames.split(',') | map('string') | join("'',''") %}
                               {% set final_variable = "''" + formatted_variable + "''" %}
                               {% set fileNameSQL = " "+"FILES =>"+""+final_variable + "," %} 
                           {%- elif  desiredState.config.fileNames  -%}
                               {% set fileNameSQL = " "+"FILES =>"+"''"+desiredState.config.fileNames+"''," %}
                           {% endif %}
                    
        {{ stage('Create inferred  '+ ' {{desiredState.node.materializationType}}') }}

           -- Anonymous stored procedure block
              DECLARE
           -- Variables
              col_name VARCHAR;
              col_type VARCHAR;
              create_sql VARCHAR;
              final_sql VARCHAR;
              alter_sql VARCHAR;
            {%if (desiredState.config.fileFormatType == 'File Format Name' and desiredState.config.fileType == 'CSV' )-%}
              file_format_options VARIANT;            
            {%-endif%}
  
                   
            -- Create statement for inferring structure
                res RESULTSET;
             {%if desiredState.config.fileFormatType == 'File Format Values' or (desiredState.config.fileFormatType == 'File Format Name' and desiredState.config.fileType == 'CSV' )-%}
                 select_statement := 'CREATE OR REPLACE {{desiredState.node.materializationType}} {{db}}.{{sch}}."{{desiredState.node.name}}"
                       USING TEMPLATE (SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
                      FROM TABLE(
                        INFER_SCHEMA(
                          LOCATION=>''@{{ dbStage }}.{{ schStage }}.{{ desiredState.config.stageName }}{{subf}}'',
                          {{fileNameSQL }}
                          FILE_FORMAT=>''"{{ dbff }}"."{{ schff }}"."TEMP_{{ desiredState.config.fileType}}_{{desiredState.node.name}}"''
                        )
                      ))';
              {%else%}
                    select_statement := 'CREATE OR REPLACE {{desiredState.node.materializationType}} {{db}}.{{sch}}."{{desiredState.node.name}}"
                       USING TEMPLATE (SELECT ARRAY_AGG(OBJECT_CONSTRUCT(*))
                      FROM TABLE(
                        INFER_SCHEMA(
                          LOCATION=>''@{{ dbStage }}.{{ schStage }}.{{ desiredState.config.stageName }}{{subf}}'',
                          {{fileNameSQL }}
                          FILE_FORMAT=>''"{{ dbff }}"."{{ schff }}"."{{ desiredState.config.fileFormatName}}"''
                        )
                      ))';
              {%endif%}
  
            BEGIN

            {%if desiredState.config.fileFormatType == 'File Format Name' and desiredState.config.fileType == 'CSV'%}
             
             -- Step 1: Run DESCRIBE to set LAST_QUERY_ID()
             EXECUTE IMMEDIATE 'DESCRIBE FILE FORMAT "{{dbff}}"."{{schff}}"."{{desiredState.config.fileFormatName}}"';


            -- Step 2: Form file format JSON string 
            SELECT PARSE_JSON(
               '{' || LISTAGG(
                 '"' || "property" || '": "' || "property_value" || '"',
                 ', '
                ) || '}'
               ) INTO file_format_options
            FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()));


              --Create temporary file format for inferring table structure 
              EXECUTE IMMEDIATE 'CREATE OR REPLACE TEMPORARY FILE FORMAT "{{dbff}}"."{{schff}}"."TEMP_{{ desiredState.config.fileType}}_{{desiredState.node.name}}"
              TYPE =''CSV''
              RECORD_DELIMITER=''' || file_format_options:RECORD_DELIMITER || '''
              FIELD_DELIMITER=''' || file_format_options:FIELD_DELIMITER || '''                                    
			  FIELD_OPTIONALLY_ENCLOSED_BY='''||file_format_options:FIELD_OPTIONALLY_ENCLOSED_BY || '''
			  BINARY_FORMAT =''' || file_format_options:BINARY_FORMAT || '''
              COMPRESSION = ''' || file_format_options:COMPRESSION || '''
              ENCODING = ''' || file_format_options:ENCODING || '''
			  PARSE_HEADER = {{desiredState.config.parsehdr1}}
              REPLACE_INVALID_CHARACTERS = TRUE
              ERROR_ON_COLUMN_COUNT_MISMATCH=FALSE
              ESCAPE = ''\\' || file_format_options:ESCAPE || '''
              ESCAPE_UNENCLOSED_FIELD = ''\\' || file_format_options:ESCAPE_UNENCLOSED_FIELD || ''' 
              TRIM_SPACE = '|| file_format_options:TRIM_SPACE ||'
              EMPTY_FIELD_AS_NULL = '|| file_format_options:EMPTY_FIELD_AS_NULL ||'
              SKIP_BYTE_ORDER_MARK = '|| file_format_options:SKIP_BYTE_ORDER_MARK ||'
              MULTI_LINE = TRUE'
              ;

            {%endif-%}
          
            {%if desiredState.config.fileFormatType == 'File Format Values'-%}
                      -- Create temporary file format
                       EXECUTE IMMEDIATE {{fileformatsql}};
            {%- endif %}
              
              
              EXECUTE IMMEDIATE :select_statement ;
	
	        -- Add metadata columns
            -- Column: FILENAME
             alter_sql := '
              ALTER TABLE {{db}}.{{sch}}."{{desiredState.node.name}}"
              ADD COLUMN FILENAME STRING';
             EXECUTE IMMEDIATE :alter_sql;

            -- Column: FILE_ROW_NUMBER
             alter_sql := '
             ALTER TABLE {{db}}.{{sch}}."{{desiredState.node.name}}"
             ADD COLUMN FILE_ROW_NUMBER NUMBER';
             EXECUTE IMMEDIATE :alter_sql;

            -- Column: FILE_LAST_MODIFIED
             alter_sql := '
             ALTER TABLE {{db}}.{{sch}}."{{desiredState.node.name}}"
             ADD COLUMN FILE_LAST_MODIFIED TIMESTAMP_NTZ';
             EXECUTE IMMEDIATE :alter_sql;

            -- Column: LOAD_TIMESTAMP
              alter_sql := '
              ALTER TABLE {{db}}.{{sch}}."{{desiredState.node.name}}"
              ADD COLUMN LOAD_TIMESTAMP TIMESTAMP_NTZ';
              EXECUTE IMMEDIATE :alter_sql;

            -- Column: SCAN_TIME
               alter_sql := '
               ALTER TABLE {{db}}.{{sch}}."{{desiredState.node.name}}"
               ADD COLUMN SCAN_TIME TIMESTAMP_NTZ';
               EXECUTE IMMEDIATE :alter_sql;
           
            {%if desiredState.config.fileFormatType == 'File Format Values' or (desiredState.config.fileFormatType == 'File Format Name' and desiredState.config.fileType == 'CSV' )-%}
                -- Drop the temporary file format
               EXECUTE IMMEDIATE 'DROP FILE FORMAT IF EXISTS "{{ dbff }}"."{{ schff }}"."TEMP_{{ desiredState.config.fileType}}_{{desiredState.node.name}}"';
            {%endif%}


            -- Return a success message
              RETURN '{{desiredState.node.name}} table created successfully.';
            END;
					  				  
	    	{%else%}
					  
					        {{ stage('Create '+ ' {{desiredState.node.materializationType}}') }}
                               CREATE OR REPLACE {{desiredState.node.materializationType}} {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
                               (
                                 {% for col in desiredState.columns %}
                                    {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, col.name) }} {{ col.dataType }}
                                 {%- if not col.nullable %} NOT NULL
                                 {%- if col.defaultValue | length > 0 %} DEFAULT {{ col.defaultValue }}{% endif %}
                                 {% endif %}
                                 {%- if col.description | length > 0 %} COMMENT '{{ col.description | escape }}'{% endif %}
                                 {%- if not loop.last -%}, {% endif %}
                                 {% endfor %}
                                 )
                                {%- if desiredState.node.description | length > 0 %} COMMENT = '{{ desiredState.node.description | escape }}'{% endif %}

		  
					 {%endif%}
        {% elif currentState != undefined and currentState.node.materializationType == desiredState.node.materializationType %}
            
            {# Desired Namespace Variables #}
            {% set nsVariables = namespace(
			colchanges=0,
			storageLocationTest=true) %}
          
           
            {# Test  #}
            {% set nsVariables.storageLocationTest = true %}
            {% set nodeCommentTest = currentState.node.description == desiredState.node.description %}
                
            {# Change Node Name or Change Storage Location #}
            {# Storage Location Tests #}
            {# Test  #}
            {%if currentState.storageLocations|length == 0%}    

              {% set currentDatabase = currentState.storageLocations | selectattr('name', 'equalto', currentState.node.location.name) | map(attribute='database') | string %}
              {% set currentSchema = currentState.storageLocations | selectattr('name', 'equalto', currentState.node.location.name) | map(attribute='schema') | string %}
               {% set currentDatabaseSchema = currentDatabase + '.' + currentSchema %}
               
            {%else %}   

               {# Current target node mappings #}
               {% set currentDatabase = currentState.storageLocations | selectattr('name', 'equalto', currentState.node.location.name) | map(attribute='database') | first | default('') %}
               {% set currentSchema = currentState.storageLocations | selectattr('name', 'equalto', currentState.node.location.name) | map(attribute='schema') | first | default('') %}
               {% set currentDatabaseSchema = currentDatabase + '.' + currentSchema %}  
			
            {% endif %}		
                
                  ## Desired mappings
                  {% set desiredDatabase = desiredState.storageLocations | selectattr('name', 'equalto', desiredState.node.location.name) | map(attribute='database') | first |default('') %}
                  {% set desiredSchema = desiredState.storageLocations | selectattr('name', 'equalto', desiredState.node.location.name) | map(attribute='schema') | first | default('')%}
            
                 {% set desiredDatabaseSchema = desiredDatabase + '.' + desiredSchema %}

                {% if currentDatabaseSchema != desiredDatabaseSchema%}
                   {% set nsVariables.storageLocationTest = false %}
                {% endif %}
           

            {# Rename / Move Object #}
            {% if (nsVariables.storageLocationTest == false or currentState.node.name != desiredState.node.name or currentState.node.location.name != desiredState.node.location.name) %}
                {{ stage('Rename and / or Move Table') }}
                ALTER TABLE {{ ref_no_link(currentState.node.location.name, currentState.node.name) }}
                RENAME TO   {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
            {% endif %}

            {# Change table description #}
            {% if (nsVariables.nodeCommentTest == false or currentState.node.description != desiredState.node.description) %}
                 {{ stage('Change table description') }}
                 ALTER TABLE {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
                SET COMMENT = '{{ desiredState.node.description | escape }}'
            {% endif %}



            {# Drop Columns #}
            
            {% for dropCurCol in currentState.columns %}
                {% if dropCurCol.id not in desiredState.columns | map(attribute="id") %}
                {% set nsVariables.colchanges = nsVariables.colchanges + 1 %}

                    {% if nsVariables.colchanges == 1 %}
                        {{ stage('Drop Columns') }}
                        ALTER TABLE {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
                        DROP COLUMN 
                    {% else %}
                        , 
                    {% endif %}
                    {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, dropCurCol.name) }} 
                {% endif %}
            {% endfor %}

            {# Add Columns #}
            
            {% for addDesCol in desiredState.columns %}
                {% if addDesCol.id not in currentState.columns | map(attribute="id") %}
                {% set nsVariables.colchanges = nsVariables.colchanges + 1 %}

                    {% if nsVariables.colchanges == 1 %}
                        {{ stage('Add Columns') }}
                        ALTER TABLE {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
                        ADD COLUMN
                    {% else %}
                        ,
                    {% endif %}
                        {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, addDesCol.name) }} {{ addDesCol.dataType }}
                        {%- if not addDesCol.nullable %} NOT NULL
                            {%- if addDesCol.defaultValue | length > 0 %} DEFAULT {{ addDesCol.defaultValue }}{% endif %}
                        {% endif %}
                        {%- if addDesCol.description | length > 0 %} COMMENT '{{ addDesCol.description | escape }}'{% endif %}
                {% endif %}
            {% endfor %}

            {# Rename Columns #}
            {% for renameCurCol in currentState.columns %}
                {% if renameCurCol.id in desiredState.columns | map(attribute="id") %}
                    {% set desiredColumnName = desiredState.columns | selectattr('id', 'equalto', renameCurCol.id) | map(attribute='name') | first %}
                    {% if renameCurCol.name != desiredColumnName %}
                        {{ stage('Rename Column ' + renameCurCol.name) }}
                        ALTER TABLE {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
                        RENAME COLUMN {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, renameCurCol.name) }}
                        TO {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, desiredColumnName) }}
                    {% endif %}
                {% endif %}
            {% endfor %}

            {# Change data types etc #}
            {% set nsVariables.colchanges = 0 %}
            {% for alterCurCol in currentState.columns %}        
                {% if alterCurCol.id in desiredState.columns | map(attribute="id") %}
                    {% set currentDatatype = alterCurCol.dataType %}
                    {% set currentNullable = alterCurCol.nullable %}
                    {% set currentDefaultValue = alterCurCol.defaultValue %}
                    {% set currentDescription = alterCurCol.description %}

                    {% set desiredColumnName = desiredState.columns | selectattr('id', "equalto", alterCurCol.id) | map(attribute='name') | first %}
                    {% set desiredDatatype = desiredState.columns | selectattr('id', "equalto", alterCurCol.id) | map(attribute='dataType') | first %}
                    {% set desiredNullable = desiredState.columns | selectattr('id', "equalto", alterCurCol.id) | map(attribute='nullable') | first %}
                    {% set desiredDefaultValue = desiredState.columns | selectattr('id', "equalto", alterCurCol.id) | map(attribute='defaultValue') | first %}
                    {% set desiredDescription = desiredState.columns | selectattr('id', "equalto", alterCurCol.id) | map(attribute='description') | first %}

                    {% set datatypeTest = currentDatatype == desiredDatatype %}
                    {% set nullableTest = currentNullable == desiredNullable %}
                    {% set defaultValueTest = currentDefaultValue == desiredDefaultValue %}
                    {% set descriptionTest = currentDescription == desiredDescription %}

                    {% if (datatypeTest == false or 
                            nullableTest == false  or
                            defaultValueTest == false or
                            descriptionTest == false) %}
                        {% set nsVariables.colchanges = nsVariables.colchanges + 1 %}

                        {% if nsVariables.colchanges == 1 %}
                            {{ stage('Change Column Attributes') }}
                            ALTER TABLE {{ ref_no_link(desiredState.node.location.name, desiredState.node.name) }}
                            MODIFY
                        {% endif %}
                        {% if datatypeTest == false %}
                            {% if nsVariables.colchanges > 1 %},{% endif %}
                            COLUMN {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, desiredColumnName) }}
                            SET DATA TYPE {{ desiredDatatype }}
                            {% set nsVariables.colchanges = nsVariables.colchanges + 1 %}
                        {% endif %}
                        {% if nullableTest == false %}
                            {% if nsVariables.colchanges > 1 %},{% endif %}
                            COLUMN {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, desiredColumnName) }}
                            {% if currentNullable == true %}
                                SET
                            {% else %}
                                DROP
                            {% endif %}
                            NOT NULL
                            {% set nsVariables.colchanges = nsVariables.colchanges + 1 %}
                        {% endif %}
                        {% if descriptionTest == false %}
                            {% if nsVariables.colchanges > 1 %},{% endif %}
                            COLUMN {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, desiredColumnName) }}
                            COMMENT '{{ desiredDescription | escape }}'
                            {% set nsVariables.colchanges = nsVariables.colchanges + 1 %}
                        {% endif %}
                        {% if defaultValueTest == false %}
                            {% if nsVariables.colchanges > 1 %},{% endif %}
                            COLUMN {{ ddlColumnCasing(desiredState.config.ddlCaseInsensitive, desiredColumnName) }}
                            {% if desiredDefaultValue == '' %}
                                DROP DEFAULT
                            {% else %}
                                SET DEFAULT {{ desiredDefaultValue }}
                            {% endif %}
                            {% set nsVariables.colchanges = nsVariables.colchanges + 1 %}
                        {% endif %}
                        
                    {% else %}
                                            
                    {% endif %}
                {% endif %}
            {% endfor %}
            
            
        {% endif %}


    {% endif %}

{% elif currentState != undefined and desiredState == undefined %}

    {# Table or View Name #}
    {% set targetObjectDatabase = ref_no_link(currentState.node.location.name, currentState.node.name).split('.')[0] %} 
    {% set targetObjectSchema = ref_no_link(currentState.node.location.name, currentState.node.name).split('.')[1] %} 
    {% set fullyQualifiedTargetObjectName = ref_no_link(currentState.node.location.name, currentState.node.name) %}
    
    {% if currentState.node.materializationType == 'view' %}

     {{ stage('Drop Table or View') }}
     DROP {{ currentState.node.materializationType }} IF EXISTS {{ fullyQualifiedTargetObjectName }}

    {% else %}
     
       {{ stage('Drop ' + currentState.node.materializationType) }}
       DROP TABLE IF EXISTS {{ fullyQualifiedTargetObjectName }}

    {% endif %}

{% endif %}