EMsnap commented on code in PR #6541:
URL: https://github.com/apache/inlong/pull/6541#discussion_r1024936947
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -205,8 +327,76 @@ public synchronized void writeRecord(T row) throws
IOException {
}
}
+ public void addSingle(T row) {
+ if (row instanceof RowData) {
+ RowData rowData = (RowData) row;
+ Map<String, String> valueMap = new HashMap<>();
+ StringJoiner value = new StringJoiner(this.fieldDelimiter);
+ for (int i = 0; i < rowData.getArity() && i < fieldGetters.length;
++i) {
+ Object field = fieldGetters[i].getFieldOrNull(rowData);
+ if (jsonFormat) {
+ String data = field != null ? field.toString() : null;
+ valueMap.put(this.fieldNames[i], data);
+ batchBytes +=
this.fieldNames[i].getBytes(StandardCharsets.UTF_8).length;
+ if (data != null) {
+ batchBytes +=
data.getBytes(StandardCharsets.UTF_8).length;
+ }
+ } else {
+ String data = field != null ? field.toString() :
NULL_VALUE;
+ value.add(data);
+ batchBytes += data.getBytes(StandardCharsets.UTF_8).length;
+ }
+ }
+ // add doris delete sign
+ if (enableBatchDelete()) {
+ if (jsonFormat) {
+ valueMap.put(DORIS_DELETE_SIGN,
parseDeleteSign(rowData.getRowKind()));
+ } else {
+ value.add(parseDeleteSign(rowData.getRowKind()));
+ }
+ }
+ Map<String,String> data = jsonFormat ? valueMap :
parsetoMap(value.toString());
+ List<Map<String, String>> mapData =
batchMap.getOrDefault(tableIdentifier, new ArrayList<String>());
+ mapData.add(data);
+ batchMap.putIfAbsent(tableIdentifier, mapData);
+ } else if (row instanceof String) {
+ batchBytes += ((String)
row).getBytes(StandardCharsets.UTF_8).length;
+ List mapData = batchMap.getOrDefault(tableIdentifier, new
ArrayList<String>());
+ mapData.add(parsetoMap(row));
+ batchMap.putIfAbsent(tableIdentifier, mapData);
+ } else {
+ throw new RuntimeException("The type of element should be
'RowData' or 'String' only.");
+ }
+ }
+
+ Map<String, String> parsetoMap(Object data) {
+ String[] toParse = data.toString().split("\\s+");
+ Map<String, String> ret = new HashMap<>();
+ if (toParse.length < 2) {
+ LOG.warn("parse length insufficient! string is :{}",
Arrays.toString(toParse));
+ return ret;
+ }
+ ret.put("id", toParse[0]);
+ ret.put("__DORIS_DELETE_SIGN__", toParse[1]);
+ return ret;
+ }
+
+ private String parseDeleteSign(RowKind rowKind) {
Review Comment:
Create a Utils for this pls
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -205,8 +327,76 @@ public synchronized void writeRecord(T row) throws
IOException {
}
}
+ public void addSingle(T row) {
+ if (row instanceof RowData) {
+ RowData rowData = (RowData) row;
+ Map<String, String> valueMap = new HashMap<>();
+ StringJoiner value = new StringJoiner(this.fieldDelimiter);
+ for (int i = 0; i < rowData.getArity() && i < fieldGetters.length;
++i) {
+ Object field = fieldGetters[i].getFieldOrNull(rowData);
+ if (jsonFormat) {
+ String data = field != null ? field.toString() : null;
+ valueMap.put(this.fieldNames[i], data);
+ batchBytes +=
this.fieldNames[i].getBytes(StandardCharsets.UTF_8).length;
+ if (data != null) {
+ batchBytes +=
data.getBytes(StandardCharsets.UTF_8).length;
+ }
+ } else {
+ String data = field != null ? field.toString() :
NULL_VALUE;
+ value.add(data);
+ batchBytes += data.getBytes(StandardCharsets.UTF_8).length;
+ }
+ }
+ // add doris delete sign
+ if (enableBatchDelete()) {
+ if (jsonFormat) {
+ valueMap.put(DORIS_DELETE_SIGN,
parseDeleteSign(rowData.getRowKind()));
+ } else {
+ value.add(parseDeleteSign(rowData.getRowKind()));
+ }
+ }
+ Map<String,String> data = jsonFormat ? valueMap :
parsetoMap(value.toString());
+ List<Map<String, String>> mapData =
batchMap.getOrDefault(tableIdentifier, new ArrayList<String>());
+ mapData.add(data);
+ batchMap.putIfAbsent(tableIdentifier, mapData);
+ } else if (row instanceof String) {
+ batchBytes += ((String)
row).getBytes(StandardCharsets.UTF_8).length;
+ List mapData = batchMap.getOrDefault(tableIdentifier, new
ArrayList<String>());
+ mapData.add(parsetoMap(row));
+ batchMap.putIfAbsent(tableIdentifier, mapData);
+ } else {
+ throw new RuntimeException("The type of element should be
'RowData' or 'String' only.");
+ }
+ }
+
+ Map<String, String> parsetoMap(Object data) {
Review Comment:
Ditto
--
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]
