fix

risingwavelabs · Jun 28, 2024 · 1981e4d · 1981e4d
1 parent 18f9980
commit 1981e4d
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/...-node/risingwave-sink-cassandra/src/main/java/com/risingwave/connector/CassandraSink.java b/...-node/risingwave-sink-cassandra/src/main/java/com/risingwave/connector/CassandraSink.java
@@ -193,7 +193,10 @@ public void drop() {
 
     private String createInsertStatement(String tableName, TableSchema tableSchema) {
         String[] columnNames = tableSchema.getColumnNames();
-        String columnNamesString = String.join(", ", columnNames);
+        String columnNamesString =
+                Arrays.stream(columnNames)
+                        .map(columnName -> "\"" + columnName + "\"")
+                        .collect(Collectors.joining(", "));
         String placeholdersString = String.join(", ", Collections.nCopies(columnNames.length, "?"));
         return String.format(
                 "INSERT INTO %s (%s) VALUES (%s)",
@@ -204,11 +207,11 @@ private String createUpdateStatement(String tableName, TableSchema tableSchema)
         List<String> primaryKeys = tableSchema.getPrimaryKeys();
         String setClause = // cassandra does not allow SET on primary keys
                 nonKeyColumns.stream()
-                        .map(columnName -> columnName + " = ?")
+                        .map(columnName -> "\"" + columnName + "\" = ?")
                         .collect(Collectors.joining(", "));
         String whereClause =
                 primaryKeys.stream()
-                        .map(columnName -> columnName + " = ?")
+                        .map(columnName -> "\"" + columnName + "\" = ?")
                         .collect(Collectors.joining(" AND "));
         return String.format("UPDATE %s SET %s WHERE %s", tableName, setClause, whereClause);
     }
@@ -217,7 +220,7 @@ private static String createDeleteStatement(String tableName, TableSchema tableS
         List<String> primaryKeys = tableSchema.getPrimaryKeys();
         String whereClause =
                 primaryKeys.stream()
-                        .map(columnName -> columnName + " = ?")
+                        .map(columnName -> "\"" + columnName + "\" = ?")
                         .collect(Collectors.joining(" AND "));
         return String.format("DELETE FROM %s WHERE %s", tableName, whereClause);
     }