diff --git a/integration_tests/starrocks-sink/README.md b/integration_tests/starrocks-sink/README.md index eb9d57ef4c26..660fcf962a6d 100644 --- a/integration_tests/starrocks-sink/README.md +++ b/integration_tests/starrocks-sink/README.md @@ -23,7 +23,7 @@ Run the following queries to create database and table. You also use other starr CREATE database demo; use demo; -CREATE table demo_bhv_table( +CREATE table upsert_table( user_id int, target_id text, event_timestamp_local datetime @@ -46,5 +46,5 @@ We only support `upsert` with starrocks' `PRIMARY KEY` Run the following query ```sql -select user_id, count(*) from demo.demo_bhv_table group by user_id; +select user_id, count(*) from demo.upsert_table group by user_id; ``` diff --git a/src/connector/src/sink/starrocks.rs b/src/connector/src/sink/starrocks.rs index c5a0740b0736..edefaf7aa120 100644 --- a/src/connector/src/sink/starrocks.rs +++ b/src/connector/src/sink/starrocks.rs @@ -139,7 +139,7 @@ impl StarrocksSink { i.name )) })?; - if !Self::check_and_correct_column_type(&i.data_type, value.to_string())? { + if !Self::check_and_correct_column_type(&i.data_type, value)? { return Err(SinkError::Starrocks(format!( "Column type don't match, column name is {:?}. starrocks type is {:?} risingwave type is {:?} ",i.name,value,i.data_type ))); @@ -150,7 +150,7 @@ impl StarrocksSink { fn check_and_correct_column_type( rw_data_type: &DataType, - starrocks_data_type: String, + starrocks_data_type: &String, ) -> Result { match rw_data_type { risingwave_common::types::DataType::Boolean => { @@ -186,12 +186,16 @@ impl StarrocksSink { risingwave_common::types::DataType::Interval => Err(SinkError::Starrocks( "INTERVAL is not supported for Starrocks sink. Please convert to VARCHAR or other supported types.".to_string(), )), - // todo! Validate the type struct and list risingwave_common::types::DataType::Struct(_) => Err(SinkError::Starrocks( "STRUCT is not supported for Starrocks sink.".to_string(), )), - risingwave_common::types::DataType::List(_) => { - Ok(starrocks_data_type.contains("unknown")) + risingwave_common::types::DataType::List(list) => { + // For compatibility with older versions starrocks + if starrocks_data_type.contains("unknown") { + return Ok(true); + } + let check_result = Self::check_and_correct_column_type(list.as_ref(), starrocks_data_type)?; + Ok(check_result && starrocks_data_type.contains("array")) } risingwave_common::types::DataType::Bytea => Err(SinkError::Starrocks( "BYTEA is not supported for Starrocks sink. Please convert to VARCHAR or other supported types.".to_string(),