fix(sink): fix starrocks sink array check (#15471)

risingwavelabs · Mar 5, 2024 · d34c7d9 · d34c7d9
1 parent d0ae778
commit d34c7d9
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 7 deletions.
diff --git 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
@@ -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<bool> {
         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(),