Issue with merge task for CDC snowflake using stream and task

Question

I have a basic task which merges data from the source table into the target table using a stream.

Creating the tables and stream

create or replace table source_json_table_trial(
v variant
);

create or replace table target_json_table_trial like source_json_table_trial;

create stream if not exists source_to_target_stream_trial on table source_json_table_trial  SHOW_INITIAL_ROWS = TRUE;

Merge Task

create or replace task stage_task_json_trial
warehouse = COMPUTE_WH
schedule = '1 minute'
when
SYSTEM$STREAM_HAS_DATA('source_to_target_stream_trial')
AS
merge into target_json_table_trial a1 using source_to_target_stream_trial b1 
on a1.v:pd:product_id = b1.v:pd:product_id 
WHEN MATCHED AND METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = 'TRUE'
then update set a1.v = b1.v, a1.lastUpdatedTimestamp= current_timestamp
WHEN NOT MATCHED AND METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = 'FALSE'
then insert values (b1.v, current_timestamp)
WHEN MATCHED AND METADATA$ACTION = 'INSERT' AND METADATA$ISUPDATE = 'FALSE'
then update set a1.v = b1.v, a1.lastUpdatedTimestamp=current_timestamp
; 

If the target has a row of id 1

INSERT INTO target_json_table_trial SELECT parse_json('{
  "pd": {
    "extraction_date": "1652787650",
    "product_id": "1",
    "product_name": "Product 1",
    "retailerName": "Retailer 1"
  }
}');    

And I insert multiple rows of same id to source at the same time

INSERT INTO source_json_table_trial SELECT parse_json('{
  "pd": {
    "extraction_date": "1652787660",
    "product_id": "1",
    "product_name": "Product 2",
    "retailerName": "Retailer 2"
  }
}'); 

INSERT INTO source_json_table_trial SELECT parse_json('{
  "pd": {
    "extraction_date": "1652787670",
    "product_id": "1",
    "product_name": "Product 3",
    "retailerName": "Retailer 3"
  }
}'); 

The new data doesn't update and is stuck in the stream.

Any ideas on what's causing this issue and how to fix it?

Answer 1

If I follow you wonderful setup sets. (but for now skip the stream to test the underlying process "works"

create or replace table source_json_table_trial(v variant);

create or replace table target_json_table_trial like source_json_table_trial;

--create stream if not exists source_to_target_stream_trial on table source_json_table_trial  SHOW_INITIAL_ROWS = TRUE;

INSERT INTO target_json_table_trial SELECT parse_json('{
  "pd": {
    "extraction_date": "1652787650",
    "product_id": "1",
    "product_name": "Product 1",
    "retailerName": "Retailer 1"
  }
}');    

INSERT INTO source_json_table_trial SELECT parse_json('{
  "pd": {
    "extraction_date": "1652787660",
    "product_id": "1",
    "product_name": "Product 2",
    "retailerName": "Retailer 2"
  }
}'); 

INSERT INTO source_json_table_trial SELECT parse_json('{
  "pd": {
    "extraction_date": "1652787670",
    "product_id": "1",
    "product_name": "Product 3",
    "retailerName": "Retailer 3"
  }
}'); 

then hand run the merge, with METADATA$ACTION = 'INSERT' replace with TRUE, and METADATA$ISUPDATE replaced with FALSE as it's not part of an update action.

merge into target_json_table_trial a1 
    using source_json_table_trial /*source_to_target_stream_trial*/ b1 
        on a1.v:pd:product_id = b1.v:pd:product_id 
WHEN MATCHED AND TRUE /*METADATA$ACTION = 'INSERT'*/ AND FALSE /*METADATA$ISUPDATE*/ = 'TRUE'
    then update 
        set a1.v = b1.v, 
            a1.lastUpdatedTimestamp= current_timestamp
WHEN NOT MATCHED AND TRUE /*METADATA$ACTION = 'INSERT'*/ AND FALSE /*METADATA$ISUPDATE*/ = 'FALSE'
    then insert values (b1.v, current_timestamp)
WHEN MATCHED AND TRUE /*METADATA$ACTION = 'INSERT'*/ AND FALSE /*METADATA$ISUPDATE*/ = 'FALSE'
    then update 
        set a1.v = b1.v, 
        a1.lastUpdatedTimestamp=current_timestamp;

I get the error:

002020 (21S01): SQL compilation error:

Insert value list does not match column list expecting 1 but got 2

which makes sense, as source_json_table_trial is just one column, and insert is:

insert values (b1.v, current_timestamp)

thus this would seem to be the problem, but I suspect this is a stripped down example. But this process of manually debugging the MERGE command should allow eliminating the MERGE as the problem, and then it becomes is the stream happy, or the task not running? Or not running when you expect..