sink emitting more rows than expected, followup of #10853

[docteurklein]

Describe the bug

followup of #10853

looks like I'm still getting extraneous sink outputs when changing only 1 element of a json structure containing siblings.

version: PostgreSQL 9.5-RisingWave-1.3.0-alpha (bb2319e)

on mysql source

update pim_catalog_product set raw_values = 
  JSON_MERGE_PATCH(raw_values, '{"desc3": {"<all_channels>": {"<all_locales>": 4}}}')
where id = 1;

expected output:

{                                                                                                                      
  "topic": "product_value_edited",                                                                                     
  "key": "{\"attribute\":\"desc3\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1}",                                                                                                                                      
  "value": "{\"attribute\":\"desc3\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1,\"value\":\"4\"}",                                                                                                                                                                                                   
  "timestamp": 1698077863141,                                                                                          
  "partition": 0,                                                                                                                                                                                                                                                                                                            
  "offset": 45                                                                                                         
}   

actual output

{                                                                                                                      
  "topic": "product_value_edited",                                                                                     
  "key": "{\"attribute\":\"desc2\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1}",               
  "timestamp": 1698077852140,                                                                                          
  "partition": 0,                                                                                                      
  "offset": 41                                                                                                         
}                                                                                                                      
{                                                                                                                      
  "topic": "product_value_edited",                                                                                     
  "key": "{\"attribute\":\"desc2\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1}",               
  "value": "{\"attribute\":\"desc2\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1,\"value\":\"4\"}",                                                                                                                                                                                                   
  "timestamp": 1698077852140,                                                                                          
  "partition": 0,                                                                                                      
  "offset": 42                                                                                                         
}                                                                                                                      
{                                                                                                                                                                                                                                             
  "topic": "product_value_edited",                                                                                     
  "key": "{\"attribute\":\"name\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1}",                
  "timestamp": 1698077863141,                                                                                                                                                                                                                                                                                                
  "partition": 0,                                                                                                      
  "offset": 43                                                                                                         
}                                                                                                                      
{                                                                                                                      
  "topic": "product_value_edited",                                                                                                                                                                                                            
  "key": "{\"attribute\":\"name\",\"channel\":null,\"locale\":null,\"product_id\":1}",                                 
  "timestamp": 1698077863141,                                                                                          
  "partition": 0,                                                                                                                                                                                                                                                                                                            
  "offset": 44                                                                                                         
}                                                                                                                      
{                                                                                                                      
  "topic": "product_value_edited",                                                                                     
  "key": "{\"attribute\":\"desc3\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1}",                                                                                                                                      
  "value": "{\"attribute\":\"desc3\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1,\"value\":\"4\"}",                                                                                                                                                                                                   
  "timestamp": 1698077863141,                                                                                          
  "partition": 0,                                                                                                                                                                                                                                                                                                            
  "offset": 45                                                                                                         
}                                                                                                                      
{                                                                                                                                                             
  "topic": "product_value_edited",                                                                                                                            
  "key": "{\"attribute\":\"name\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1}",                                                       
  "value": "{\"attribute\":\"name\",\"channel\":null,\"locale\":\"\u003call_locales\",\"product_id\":1,\"value\":\"{}\"}",                                                                                                                                                                                                   
  "timestamp": 1698077863141,                                                                                                                                 
  "partition": 0,                                                                                                                                             
  "offset": 46                                                                                                                                                
}                                                                                                                                                             
{                                                                                                                                                             
  "topic": "product_value_edited",                                                                                                                            
  "key": "{\"attribute\":\"name\",\"channel\":null,\"locale\":null,\"product_id\":1}",                                                                        
  "value": "{\"attribute\":\"name\",\"channel\":null,\"locale\":null,\"product_id\":1,\"value\":\"0.4290362600507552\"}",                                                                                                                                                                                                    
  "timestamp": 1698077863141,                                                                                                                                 
  "partition": 0,                                                                                                                                             
  "offset": 47    
}

[st1page]

I guess the key is expected to be pim_catalog_product.id instead of the full raw_values.

Would you please provide the complete CREATE SINK statement and the explain result of it?

explain CREATE SINK sink_name  ....

Also the table's definition in the downstream PG.

[docteurklein]

for sure, sorry for not including it (I've got the source code here just in case: https://github.com/docteurklein/risingwave-demo/blob/main/rw/product.sql)

sink DDL

create sink product_value_edited from pim1.product_value
with (
    connector='kafka',
    type='upsert',
    primary_key='product_id, attribute, channel, locale',
    -- force_append_only='true',
    properties.bootstrap.server='redpanda:9092',
    topic='product_value_edited'
);

explain create sink

 StreamSink { type: upsert, columns: [product_id, attribute, channel, locale, value, product_value.product.uuid(hidden), product_value.projected_row_id(hidden), product_value.projected_row_id#1(hidden), product_value.projected_row_id#2(hidden)], pk: [product_value.product.uuid, product_value.projected_row_id, product_value.projected_row_id#1, product_value.projected_row_id#2] }
 └─StreamTableScan { table: product_value, columns: [product_id, attribute, channel, locale, value, product.uuid, projected_row_id, projected_row_id#1, projected_row_id#2] }

mysql source DDL

create table pim_catalog_product (
  uuid binary(16),
  family_id int,
  product_model_id int,
  family_variant_id int,
  id int,
  is_enabled bool,
  identifier text,
  raw_values json,
  created timestamp,
  updated timestamp,
  primary key (uuid)
);

insert into pim_catalog_product (uuid, id, family_id, raw_values) values
(UUID_TO_BIN(uuid()), 1, 1, '{}');

I'll gladly provide anything that could help, Thanks for the quick answer!

[st1page]

I guess you should change the primary key to primary_key='product_id' ? the actual output exactly output records with different pks. 🤔 Can you provide the result of the SELECT * FROM pim1.product_value?

[stdrc]

Looks like that the update command will merge "desc3": ... into raw_values which may be {"name": ..., "desc2": ...}, hence, this UPDATE appear to RisingWave as a whole row update.

Then, during jsonb_each, RW expands the row update to multiple delete and insert, in this process, RW cannot easily merge unchanged "name" and "desc2" rows due to our internal mechanism. This finally leads to multiple update records on sink.

We may support update compaction by sink pk when sinking in the future (still investigating), but currently it seems to be expected behavior (as long as the final output is correct). If the final result is actually wrong, please notify us, may need further debugging.

[docteurklein]

Sorry, no computer access right now, but yes that's exactly what @stdrc described. The output always contains correct values (although some are extraneous since some rows of the mat view didn't change (like name and desc2 in the example).

Ok so I'll wait for update compaction then. Thanks for the clarifications!

[st1page]

I think it has been fixed after #15345 and ok in release v1.8

[stdrc]

I think it has been fixed after #15345 and ok in release v1.8

Closing this issue since the fix has been merged.