Hisoka-X commented on code in PR #7888:
URL: https://github.com/apache/seatunnel/pull/7888#discussion_r1817668817
##########
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java:
##########
@@ -60,23 +68,94 @@ public RedisSinkWriter(SeaTunnelRowType seaTunnelRowType,
RedisParameters redisP
@Override
public void write(SeaTunnelRow element) throws IOException {
- String data = new String(serializationSchema.serialize(element));
- String keyField = redisParameters.getKeyField();
List<String> fields = Arrays.asList(seaTunnelRowType.getFieldNames());
- String key;
- if (fields.contains(keyField)) {
- key = element.getField(fields.indexOf(keyField)).toString();
- } else {
- key = keyField;
- }
+ String key = getKey(element, fields);
keyBuffer.add(key);
- valueBuffer.add(data);
+ String value = getValue(element, fields);
+ valueBuffer.add(value);
if (keyBuffer.size() >= batchSize) {
doBatchWrite();
clearBuffer();
}
}
+ private String getKey(SeaTunnelRow element, List<String> fields) {
+ String keyField = redisParameters.getKeyField();
+ String[] keyFieldSegments = keyField.split(REDIS_GROUP_DELIMITER);
+ StringBuilder key = new StringBuilder();
+ for (int i = 0; i < keyFieldSegments.length; i++) {
+ String keyFieldSegment = keyFieldSegments[i];
+ if (keyFieldSegment.startsWith(LEFT_PLACEHOLDER_MARKER)
+ && keyFieldSegment.endsWith(RIGHT_PLACEHOLDER_MARKER)) {
+ String realKeyField = keyFieldSegment.substring(1,
keyFieldSegment.length() - 1);
+ if (fields.contains(realKeyField)) {
+
key.append(element.getField(fields.indexOf(realKeyField)).toString());
+ } else {
+ key.append(keyFieldSegment);
+ }
+ } else {
+ key.append(keyFieldSegment);
+ }
+ if (i != keyFieldSegments.length - 1) {
+ key.append(REDIS_GROUP_DELIMITER);
+ }
+ }
+ return key.toString();
+ }
+
+ private String getValue(SeaTunnelRow element, List<String> fields) {
+ String value;
+ RedisDataType redisDataType = redisParameters.getRedisDataType();
+ if (RedisDataType.HASH.equals(redisDataType)) {
+ value = handleHashType(element, fields);
+ } else {
+ value = handleOtherTypes(element, fields);
+ }
+ if (StringUtils.isEmpty(value)) {
+ byte[] serialize = serializationSchema.serialize(element);
+ value = new String(serialize);
+ }
Review Comment:
If the value of the data is originally "", the current logic will cause data
abnormality. Do I understand it correctly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
