Handle more than 100 fields to compute hashid

airbyte-integrations/bases/base-normalization/Dockerfile

@@ -16,5 +16,5 @@ WORKDIR /airbyte
 
 ENTRYPOINT ["/airbyte/entrypoint.sh"]
 
-LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.version=0.1.1
 LABEL io.airbyte.name=airbyte/normalization

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/concat.sql

@@ -0,0 +1,33 @@
+{#
+    Overriding the following macro from dbt-utils:
+        https://github.com/fishtown-analytics/dbt-utils/blob/0.6.2/macros/cross_db_utils/concat.sql
+    To implement our own version of concat
+    Because on postgres, we cannot pass more than 100 arguments to a function
+#}
+
+{% macro concat(fields) -%}
+  {{ adapter.dispatch('concat')(fields) }}
+{%- endmacro %}
+
+{% macro default__concat(fields) -%}
+    concat({{ fields|join(', ') }})
+{%- endmacro %}
+
+{% macro alternative_concat(fields) %}
+    {{ fields|join(' || ') }}
+{% endmacro %}
+
+
+{% macro postgres__concat(fields) %}
+    {{ dbt_utils.alternative_concat(fields) }}
+{% endmacro %}
+
+
+{% macro redshift__concat(fields) %}
+    {{ dbt_utils.alternative_concat(fields) }}
+{% endmacro %}
+
+
+{% macro snowflake__concat(fields) %}
+    {{ dbt_utils.alternative_concat(fields) }}
+{% endmacro %}

airbyte-integrations/bases/base-normalization/dbt-project-template/macros/cross_db_utils/surrogate_key.sql

@@ -0,0 +1,51 @@
+{#
+    Overriding the following macro from dbt-utils:
+        https://github.com/fishtown-analytics/dbt-utils/blob/0.6.2/macros/sql/surrogate_key.sql
+    To implement our own version of concat
+    Because on postgres, we cannot pass more than 100 arguments to a function
+#}
+
+{%- macro surrogate_key(field_list) -%}
+
+{%- if varargs|length >= 1 or field_list is string %}
+
+{%- set error_message = '
+Warning: the `surrogate_key` macro now takes a single list argument instead of \
+multiple string arguments. Support for multiple string arguments will be \
+deprecated in a future release of dbt-utils. The {}.{} model triggered this warning. \
+'.format(model.package_name, model.name) -%}
+
+{%- do exceptions.warn(error_message) -%}
+
+{# first argument is not included in varargs, so add first element to field_list_xf #}
+{%- set field_list_xf = [field_list] -%}
+
+{%- for field in varargs %}
+{%- set _ = field_list_xf.append(field) -%}
+{%- endfor -%}
+
+{%- else -%}
+
+{# if using list, just set field_list_xf as field_list #}
+{%- set field_list_xf = field_list -%}
+
+{%- endif -%}
+
+
+{%- set fields = [] -%}
+
+{%- for field in field_list_xf -%}
+
+    {%- set _ = fields.append(
+        "coalesce(cast(" ~ field ~ " as " ~ dbt_utils.type_string() ~ "), '')"
+    ) -%}
+
+    {%- if not loop.last %}
+        {%- set _ = fields.append("'-'") -%}
+    {%- endif -%}
+
+{%- endfor -%}
+
+{{dbt_utils.hash(concat(fields))}}
+
+{%- endmacro -%}

airbyte-integrations/bases/base-normalization/normalization/transform_catalog/transform.py

@@ -299,7 +299,10 @@ def process_node(
     node_properties = extract_node_properties(path=path, json_col=json_col, properties=properties)
     node_columns = ",\n    ".join([sql for sql in node_properties.values()])
     hash_node_columns = ",\n        ".join([f"adapter.quote_as_configured('{column}', 'identifier')" for column in node_properties.keys()])
-    hash_node_columns = jinja_call(f"dbt_utils.surrogate_key([\n        {hash_node_columns}\n    ])")
+    # Disable dbt_utils.surrogate_key for own version to fix a bug with Postgres (#913).
+    # hash_node_columns = jinja_call(f"dbt_utils.surrogate_key([\n        {hash_node_columns}\n    ])")
+    # We should re-enable it when our PR to dbt_utils is merged
+    hash_node_columns = jinja_call(f"surrogate_key([\n        {hash_node_columns}\n    ])")
     hash_id = jinja_call(f"adapter.quote_as_configured('_{name}_hashid', 'identifier')")
     foreign_hash_id = jinja_call(f"adapter.quote_as_configured('_{name}_foreign_hashid', 'identifier')")
     emitted_col = "{},\n    {} as {}".format(

airbyte-workers/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java

@@ -43,7 +43,7 @@ public class DefaultNormalizationRunner implements NormalizationRunner {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DefaultNormalizationRunner.class);
 
-  public static final String NORMALIZATION_IMAGE_NAME = "airbyte/normalization:0.1.0";
+  public static final String NORMALIZATION_IMAGE_NAME = "airbyte/normalization:0.1.1";
 
   private final DestinationType destinationType;
   private final ProcessBuilderFactory pbf;