🐛 Fix normalization SCD partition by float columns errors with BigQuery (#9281)

Author: ChristopheDuong

airbyte-integrations/bases/base-normalization/integration_tests/normalization_test_output/bigquery/test_nested_streams/first_output/airbyte_incremental/scd/test_normalization/nested_stream_with_complex_columns_resulting_into_long_names_scd.sql

@@ -26,9 +26,9 @@ scd_data as (
 ), '')) as 
     string
 ))) as _airbyte_unique_key,
-        id,
-        date,
-        `partition`,
+      id,
+      date,
+      `partition`,
       date as _airbyte_start_at,
       lag(date) over (
         partition by id
@@ -54,7 +54,10 @@ dedup_data as (
         -- we need to ensure de-duplicated rows for merge/update queries
         -- additionally, we generate a unique key for the scd table
         row_number() over (
-            partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
+            partition by
+                _airbyte_unique_key,
+                _airbyte_start_at,
+                _airbyte_emitted_at
             order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         to_hex(md5(cast(concat(coalesce(cast(_airbyte_unique_key as 
@@ -72,9 +75,9 @@ dedup_data as (
 select
     _airbyte_unique_key,
     _airbyte_unique_key_scd,
-        id,
-        date,
-        `partition`,
+    id,
+    date,
+    `partition`,
     _airbyte_start_at,
     _airbyte_end_at,
     _airbyte_active_row,

airbyte-integrations/bases/base-normalization/integration_tests/normalization_test_output/bigquery/test_nested_streams/models/generated/airbyte_incremental/scd/test_normalization/nested_stream_with_complex_columns_resulting_into_long_names_scd.sql

@@ -57,11 +57,11 @@ scd_data as (
     -- SQL model to build a Type 2 Slowly Changing Dimension (SCD) table for each record identified by their primary key
     select
       {{ dbt_utils.surrogate_key([
-            'id',
+      'id',
       ]) }} as _airbyte_unique_key,
-        id,
-        date,
-        {{ adapter.quote('partition') }},
+      id,
+      date,
+      {{ adapter.quote('partition') }},
       date as _airbyte_start_at,
       lag(date) over (
         partition by id
@@ -87,7 +87,10 @@ dedup_data as (
         -- we need to ensure de-duplicated rows for merge/update queries
         -- additionally, we generate a unique key for the scd table
         row_number() over (
-            partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
+            partition by
+                _airbyte_unique_key,
+                _airbyte_start_at,
+                _airbyte_emitted_at
             order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         {{ dbt_utils.surrogate_key([
@@ -101,9 +104,9 @@ dedup_data as (
 select
     _airbyte_unique_key,
     _airbyte_unique_key_scd,
-        id,
-        date,
-        {{ adapter.quote('partition') }},
+    id,
+    date,
+    {{ adapter.quote('partition') }},
     _airbyte_start_at,
     _airbyte_end_at,
     _airbyte_active_row,

airbyte-integrations/bases/base-normalization/integration_tests/normalization_test_output/bigquery/test_simple_streams/first_output/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql

@@ -30,14 +30,14 @@ scd_data as (
 ), '')) as 
     string
 ))) as _airbyte_unique_key,
-        id,
-        currency,
-        date,
-        timestamp_col,
-        HKD_special___characters,
-        HKD_special___characters_1,
-        NZD,
-        USD,
+      id,
+      currency,
+      date,
+      timestamp_col,
+      HKD_special___characters,
+      HKD_special___characters_1,
+      NZD,
+      USD,
       date as _airbyte_start_at,
       lag(date) over (
         partition by id, currency, cast(NZD as 
@@ -67,7 +67,10 @@ dedup_data as (
         -- we need to ensure de-duplicated rows for merge/update queries
         -- additionally, we generate a unique key for the scd table
         row_number() over (
-            partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
+            partition by
+                _airbyte_unique_key,
+                _airbyte_start_at,
+                _airbyte_emitted_at
             order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         to_hex(md5(cast(concat(coalesce(cast(_airbyte_unique_key as 
@@ -85,14 +88,14 @@ dedup_data as (
 select
     _airbyte_unique_key,
     _airbyte_unique_key_scd,
-        id,
-        currency,
-        date,
-        timestamp_col,
-        HKD_special___characters,
-        HKD_special___characters_1,
-        NZD,
-        USD,
+    id,
+    currency,
+    date,
+    timestamp_col,
+    HKD_special___characters,
+    HKD_special___characters_1,
+    NZD,
+    USD,
     _airbyte_start_at,
     _airbyte_end_at,
     _airbyte_active_row,

airbyte-integrations/bases/base-normalization/integration_tests/normalization_test_output/bigquery/test_simple_streams/models/generated/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql

@@ -59,18 +59,18 @@ scd_data as (
     -- SQL model to build a Type 2 Slowly Changing Dimension (SCD) table for each record identified by their primary key
     select
       {{ dbt_utils.surrogate_key([
-            'id',
-            'currency',
-            'NZD',
+      'id',
+      'currency',
+      'NZD',
       ]) }} as _airbyte_unique_key,
-        id,
-        currency,
-        date,
-        timestamp_col,
-        HKD_special___characters,
-        HKD_special___characters_1,
-        NZD,
-        USD,
+      id,
+      currency,
+      date,
+      timestamp_col,
+      HKD_special___characters,
+      HKD_special___characters_1,
+      NZD,
+      USD,
       date as _airbyte_start_at,
       lag(date) over (
         partition by id, currency, cast(NZD as {{ dbt_utils.type_string() }})
@@ -96,7 +96,10 @@ dedup_data as (
         -- we need to ensure de-duplicated rows for merge/update queries
         -- additionally, we generate a unique key for the scd table
         row_number() over (
-            partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
+            partition by
+                _airbyte_unique_key,
+                _airbyte_start_at,
+                _airbyte_emitted_at
             order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         {{ dbt_utils.surrogate_key([
@@ -110,14 +113,14 @@ dedup_data as (
 select
     _airbyte_unique_key,
     _airbyte_unique_key_scd,
-        id,
-        currency,
-        date,
-        timestamp_col,
-        HKD_special___characters,
-        HKD_special___characters_1,
-        NZD,
-        USD,
+    id,
+    currency,
+    date,
+    timestamp_col,
+    HKD_special___characters,
+    HKD_special___characters_1,
+    NZD,
+    USD,
     _airbyte_start_at,
     _airbyte_end_at,
     _airbyte_active_row,

airbyte-integrations/bases/base-normalization/integration_tests/normalization_test_output/bigquery/test_simple_streams/modified_models/generated/airbyte_incremental/scd/test_normalization/dedup_exchange_rate_scd.sql

@@ -59,18 +59,18 @@ scd_data as (
     -- SQL model to build a Type 2 Slowly Changing Dimension (SCD) table for each record identified by their primary key
     select
       {{ dbt_utils.surrogate_key([
-            'id',
-            'currency',
-            'NZD',
+      'id',
+      'currency',
+      'NZD',
       ]) }} as _airbyte_unique_key,
-        id,
-        currency,
-        new_column,
-        date,
-        timestamp_col,
-        HKD_special___characters,
-        NZD,
-        USD,
+      id,
+      currency,
+      new_column,
+      date,
+      timestamp_col,
+      HKD_special___characters,
+      NZD,
+      USD,
       date as _airbyte_start_at,
       lag(date) over (
         partition by cast(id as {{ dbt_utils.type_string() }}), currency, cast(NZD as {{ dbt_utils.type_string() }})
@@ -96,7 +96,10 @@ dedup_data as (
         -- we need to ensure de-duplicated rows for merge/update queries
         -- additionally, we generate a unique key for the scd table
         row_number() over (
-            partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at
+            partition by
+                _airbyte_unique_key,
+                _airbyte_start_at,
+                _airbyte_emitted_at
             order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         {{ dbt_utils.surrogate_key([
@@ -110,14 +113,14 @@ dedup_data as (
 select
     _airbyte_unique_key,
     _airbyte_unique_key_scd,
-        id,
-        currency,
-        new_column,
-        date,
-        timestamp_col,
-        HKD_special___characters,
-        NZD,
-        USD,
+    id,
+    currency,
+    new_column,
+    date,
+    timestamp_col,
+    HKD_special___characters,
+    NZD,
+    USD,
     _airbyte_start_at,
     _airbyte_end_at,
     _airbyte_active_row,

airbyte-integrations/bases/base-normalization/integration_tests/normalization_test_output/clickhouse/test_simple_streams/first_output/airbyte_incremental/scd/test_normalization/dedup_cdc_excluded_scd.sql

@@ -26,9 +26,10 @@ input_data_with_active_row_num as (
       row_number() over (
         partition by id
         order by
-            _airbyte_emitted_at is null asc,
-            _airbyte_emitted_at desc,
-            _airbyte_emitted_at desc, _ab_cdc_updated_at desc
+            _ab_cdc_lsn is null asc,
+            _ab_cdc_lsn desc,
+            _ab_cdc_updated_at desc,
+            _airbyte_emitted_at desc
       ) as _airbyte_active_row_num
     from input_data
 ),
@@ -40,21 +41,21 @@ scd_data as (
                 toString(id)
 
     ))) as _airbyte_unique_key,
-        id,
-        name,
-        _ab_cdc_lsn,
-        _ab_cdc_updated_at,
-        _ab_cdc_deleted_at,
-      _airbyte_emitted_at as _airbyte_start_at,
+      id,
+      name,
+      _ab_cdc_lsn,
+      _ab_cdc_updated_at,
+      _ab_cdc_deleted_at,
+      _ab_cdc_lsn as _airbyte_start_at,
       case when _airbyte_active_row_num = 1 and _ab_cdc_deleted_at is null then 1 else 0 end as _airbyte_active_row,
-      anyOrNull(_airbyte_emitted_at) over (
+      anyOrNull(_ab_cdc_lsn) over (
         partition by id
         order by
-            _airbyte_emitted_at is null asc,
-            _airbyte_emitted_at desc,
-            _airbyte_emitted_at desc, _ab_cdc_updated_at desc
-            ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
-      ) as _airbyte_end_at,
+            _ab_cdc_lsn is null asc,
+            _ab_cdc_lsn desc,
+            _ab_cdc_updated_at desc,
+            _airbyte_emitted_at desc
+            ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING) as _airbyte_end_at,
       _airbyte_ab_id,
       _airbyte_emitted_at,
       _airbyte_dedup_cdc_excluded_hashid
@@ -65,7 +66,10 @@ dedup_data as (
         -- we need to ensure de-duplicated rows for merge/update queries
         -- additionally, we generate a unique key for the scd table
         row_number() over (
-            partition by _airbyte_unique_key, _airbyte_start_at, _airbyte_emitted_at, accurateCastOrNull(_ab_cdc_deleted_at, 'String'), accurateCastOrNull(_ab_cdc_updated_at, 'String')
+            partition by
+                _airbyte_unique_key,
+                _airbyte_start_at,
+                _airbyte_emitted_at, accurateCastOrNull(_ab_cdc_deleted_at, 'String'), accurateCastOrNull(_ab_cdc_updated_at, 'String')
             order by _airbyte_active_row desc, _airbyte_ab_id
         ) as _airbyte_row_num,
         assumeNotNull(hex(MD5(
@@ -91,11 +95,11 @@ dedup_data as (
 select
     _airbyte_unique_key,
     _airbyte_unique_key_scd,
-        id,
-        name,
-        _ab_cdc_lsn,
-        _ab_cdc_updated_at,
-        _ab_cdc_deleted_at,
+    id,
+    name,
+    _ab_cdc_lsn,
+    _ab_cdc_updated_at,
+    _ab_cdc_deleted_at,
     _airbyte_start_at,
     _airbyte_end_at,
     _airbyte_active_row,