auto add __op for primary key table even when user specified `sink.…

…properties.columns`
StarRocks · Mar 27, 2024 · db21f99 · db21f99
1 parent 697cb58
commit db21f99
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/src/main/java/com/starrocks/connector/flink/table/sink/StarRocksSinkOptions.java b/src/main/java/com/starrocks/connector/flink/table/sink/StarRocksSinkOptions.java
@@ -514,15 +514,15 @@ public StreamLoadProperties getProperties(@Nullable StarRocksSinkTable table) {
                 .enableUpsertDelete(supportUpsertDelete());
 
         if (hasColumnMappingProperty()) {
-            String columns = streamLoadProps.get("columns");
+            List<String> columns = Arrays.asList(streamLoadProps.get("columns").split(","));
             if (supportUpsertDelete()) {
                 // auto add `__op` for primary key table even when user specified `sink.properties.columns`.
                 // in case user use a bitmap datatype and need set up `sink.properties.columns`, may forget to add `__op`.
-                if (!columns.endsWith(",__op")) {
-                    columns = columns + ",__op";
+                if (columns.stream().noneMatch(it -> it.equals("__op"))) {
+                    columns.add("__op");
                 }
             }
-            defaultTablePropertiesBuilder.columns(columns);
+            defaultTablePropertiesBuilder.columns(String.join(",", columns));
         } else if (getTableSchemaFieldNames() != null) {
             // don't need to add "columns" header in following cases
             // 1. use csv format but the flink and starrocks schemas are aligned