rows_affected in adapter response is 0 for incremental models

[unstoppable-allensun]

I seem to be having issues since upgrading to 1.8.4+ of the sql server adapter where the rows_affected is coming back as 0 for incremental models. It used to work in 1.4.3. Is anyone else having this issue or able to reproduce it?

[cody-scott]

If you pin sql server to 1.8.4 and fabric to 1.8.7 does it work?

[unstoppable-allensun]

No, that’s what I had originally and then upgraded to 1.8.5 and 1.8.8 of fabric to see if that fixed anything. The issue seems to be happening on the first run when inserting new records into incremental table but when I run again with the same condition it will report correctly.

[unstoppable-allensun]

@cody-scott Also, for table materialization, the rows_affected is always -1 for me.

[unstoppable-allensun]

So I was playing around the fabric__get_delete_insert_merge_sql macro and it seems that if I don't have a unique_key, the cursor properly reports the rowcount. It seems the rowcount being reported is from the delete statement and the insert doesn't change the rowcount. Was this maybe a change in pyodbc functionality not reporting last statement rowcount?

{% macro fabric__get_delete_insert_merge_sql(target, source, unique_key, dest_columns, incremental_predicates=none) %}

    {% set query_label = apply_label() %}
    {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}

    {% if unique_key %}
        {% if unique_key is sequence and unique_key is not string %}
            delete from {{ target }}
            where exists (
                select null
                from {{ source }}
                where
                {% for key in unique_key %}
                    {{ source }}.{{ key }} = {{ target }}.{{ key }}
                    {{ "and " if not loop.last }}
                {% endfor %}
            )
            {% if incremental_predicates %}
                {% for predicate in incremental_predicates %}
                    and {{ predicate }}
                {% endfor %}
            {% endif %}
            {{ query_label }}
        {% else %}
            delete from {{ target }}
            where (
                {{ unique_key }}) in (
                select ({{ unique_key }})
                from {{ source }}
            )
            {%- if incremental_predicates %}
                {% for predicate in incremental_predicates %}
                    and {{ predicate }}
                {% endfor %}
            {%- endif -%}
            {{ query_label }}
        {% endif %}
    {% endif %}

    insert into {{ target }} ({{ dest_cols_csv }})
    (
        select {{ dest_cols_csv }}
        from {{ source }}
    ){{ query_label }}
{% endmacro %}

[unstoppable-allensun]

So wrapping SET NOCOUNT ON; and SET NOCOUNT OFF; around the DELETE statements resolved the issue for me.

[unstoppable-allensun]

I was also able to fix the rowcount for table materialization as well by calling the create view in a separate batch.

{% macro sqlserver__create_table_as(temporary, relation, sql) -%}
    {%- set query_label = apply_label() -%}
    {%- set tmp_relation = relation.incorporate(path={"identifier": relation.identifier ~ '__dbt_tmp_vw'}, type='view') -%}

    {%- do adapter.drop_relation(tmp_relation) -%}

    {% call statement('create_temp_view', fetch_result=False) %}
        USE [{{ relation.database }}];
        {{ get_create_view_as_sql(tmp_relation, sql) }}
    {% endcall %}

    {%- set table_name -%}
        {{ relation }}
    {%- endset -%}


    {%- set contract_config = config.get('contract') -%}
    {%- set query -%}
        {% if contract_config.enforced and (not temporary) %}
            CREATE TABLE {{table_name}}
            {{ get_assert_columns_equivalent(sql)  }}
            {{ build_columns_constraints(relation) }}
            {% set listColumns %}
                {% for column in model['columns'] %}
                    {{ "["~column~"]" }}{{ ", " if not loop.last }}
                {% endfor %}
            {%endset%}
            INSERT INTO {{relation}} ({{listColumns}})
            SELECT {{listColumns}} FROM {{tmp_relation}} {{ query_label }}

        {% else %}
            SELECT * INTO {{ table_name }} FROM {{ tmp_relation }} {{ query_label }}
        {% endif %}
    {%- endset -%}

    EXEC('{{- escape_single_quotes(query) -}}')

    {# For some reason drop_relation is not firing. This solves the issue for now. #}
    EXEC('DROP VIEW IF EXISTS {{tmp_relation.schema}}.{{tmp_relation.identifier}}')



    {% set as_columnstore = config.get('as_columnstore', default=true) %}
    {% if not temporary and as_columnstore -%}
        {#-
        add columnstore index
        this creates with dbt_temp as its coming from a temporary relation before renaming
        could alter relation to drop the dbt_temp portion if needed
        -#}
        {{ sqlserver__create_clustered_columnstore_index(relation) }}
   {% endif %}

{% endmacro %}

[cody-scott]

@unstoppable-allensun would you be able to open a PR with your proposed solution?