Fix: normalize rows_affected to handle string values from dbt-databricks (#915)

* Fix: normalize rows_affected to handle string values from dbt-databricks

When using dbt-databricks with view_update_via_alter enabled, skipped views
return rows_affected as a string '-1' instead of an integer. This causes
Spark's VALUES clause to fail with INVALID_INLINE_TABLE.INCOMPATIBLE_TYPES_IN_INLINE_TABLE
when building bulk INSERTs for dbt_run_results.

The fix adds a normalize_rows_affected macro that:
- Converts string '-1' to null (skipped operations have no meaningful row count)
- Converts other string numbers to integers (defensive handling)
- Passes through normal integers unchanged

Fixes: elementary-data/elementary#2089
Linear: CORE-267
Co-Authored-By: Yosef Arbiv <yosef.arbiv@gmail.com>

* Add unit test for normalize_rows_affected macro

Tests the following cases:
- None input returns None
- Integer inputs pass through unchanged (123, 0, -1)
- String '-1' returns None (skipped operations)
- Other string numbers are converted to integers ('123', '0', '456')

Co-Authored-By: Yosef Arbiv <yosef.arbiv@gmail.com>

* Fix test macro to not double-encode JSON results

The log_macro_results helper already applies tojson() to the result,
so the test macro should return the raw value instead of tojson(result).

Co-Authored-By: Yosef Arbiv <yosef.arbiv@gmail.com>

* Fix test to handle None return values from macro

When the macro returns None, log_macro_results doesn't log anything,
so run_operation returns an empty list. The test now handles this case
by checking if the result list is empty before parsing.

Co-Authored-By: Yosef Arbiv <yosef.arbiv@gmail.com>

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Yosef Arbiv <yosef.arbiv@gmail.com>

Author: devin-ai-integration[bot]

integration_tests/dbt_project/macros/test_normalize_rows_affected.sql

@@ -0,0 +1,4 @@
+{% macro test_normalize_rows_affected(rows_affected) %}
+    {% set result = elementary.normalize_rows_affected(rows_affected) %}
+    {{ return(result) }}
+{% endmacro %}

integration_tests/tests/test_dbt_artifacts/test_normalize_rows_affected.py

@@ -0,0 +1,34 @@
+import json
+
+import pytest
+from dbt_project import DbtProject
+
+
+@pytest.mark.parametrize(
+    "input_value,expected_output",
+    [
+        (None, None),
+        (123, 123),
+        (0, 0),
+        (-1, -1),
+        ("123", 123),
+        ("0", 0),
+        ("-1", None),
+        ("456", 456),
+    ],
+)
+def test_normalize_rows_affected(dbt_project: DbtProject, input_value, expected_output):
+    result = dbt_project.dbt_runner.run_operation(
+        "elementary_tests.test_normalize_rows_affected",
+        macro_args={"rows_affected": input_value},
+    )
+    # When the macro returns None, log_macro_results doesn't log anything,
+    # so run_operation returns an empty list
+    if not result:
+        actual_output = None
+    else:
+        actual_output = json.loads(result[0])
+    assert actual_output == expected_output, (
+        f"normalize_rows_affected({input_value!r}) returned {actual_output!r}, "
+        f"expected {expected_output!r}"
+    )

macros/edr/dbt_artifacts/upload_run_results.sql

@@ -37,18 +37,33 @@
     {{ return(dbt_run_results_empty_table_query) }}
 {% endmacro %}
 
+{% macro normalize_rows_affected(rows_affected) %}
+    {% if rows_affected is none %}
+        {{ return(none) }}
+    {% elif rows_affected is string %}
+        {% if rows_affected == '-1' %}
+            {{ return(none) }}
+        {% else %}
+            {{ return(rows_affected | int) }}
+        {% endif %}
+    {% else %}
+        {{ return(rows_affected) }}
+    {% endif %}
+{% endmacro %}
+
 {% macro flatten_run_result(run_result) %}
     {% set run_result_dict = elementary.get_run_result_dict(run_result) %}
     {% set node = elementary.safe_get_with_default(run_result_dict, 'node', {}) %}
     {% set config_dict = elementary.safe_get_with_default(node, 'config', {}) %}
+    {% set raw_rows_affected = run_result_dict.get('adapter_response', {}).get('rows_affected') %}
     {% set flatten_run_result_dict = {
         'model_execution_id': elementary.get_node_execution_id(node),
         'invocation_id': invocation_id,
         'unique_id': node.get('unique_id'),
         'name': node.get('name'),
         'message': run_result_dict.get('message'),
         'generated_at': elementary.datetime_now_utc_as_string(),
-        'rows_affected': run_result_dict.get('adapter_response', {}).get('rows_affected'),
+        'rows_affected': elementary.normalize_rows_affected(raw_rows_affected),
         'execution_time': run_result_dict.get('execution_time'),
         'status': run_result_dict.get('status'),
         'resource_type': node.get('resource_type'),