yunqingmoswu commented on code in PR #6541:
URL: https://github.com/apache/inlong/pull/6541#discussion_r1024790515
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicTableSink.java:
##########
@@ -87,16 +85,20 @@ public ChangelogMode getChangelogMode(ChangelogMode
changelogMode) {
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
if (!multipleSink) {
Review Comment:
This piece of code redundancy can be merged
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -198,15 +320,92 @@ private boolean checkFlushException(String
tableIdentifier) {
@Override
public synchronized void writeRecord(T row) throws IOException {
addBatch(row);
+ if (isSingle) {
+ size = batch.size();
+ }
boolean valid = (executionOptions.getBatchSize() > 0 && size >=
executionOptions.getBatchSize())
|| batchBytes >= executionOptions.getMaxBatchBytes();
if (valid && !flushing) {
flush();
}
}
+ 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()));
+ }
+ }
+ Object data = jsonFormat ? valueMap : value.toString();
+ LOG.info("appending data object {} with jsonformat {}", data,
jsonFormat);
+ LOG.info("parsed data {}", parsetoMap(data));
+ //appending data object 12345 0
+ //should be {"id":"12345","__DORIS_DELETE_SIGN__":"0"}
+ List<Map<String, String>> mapData =
batchMap.getOrDefault(tableIdentifier, new ArrayList<String>());
+ mapData.add(parsetoMap(data));
+ batchMap.putIfAbsent(tableIdentifier, mapData);
+ } else if (row instanceof String) {
+ batchBytes += ((String)
row).getBytes(StandardCharsets.UTF_8).length;
+ LOG.info("appending row {}", row);
+ 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;
+ }
+ LOG.info("String to parse id: {} delete: {}", toParse[0], toParse[1]);
Review Comment:
Log printing too often？
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -198,15 +320,92 @@ private boolean checkFlushException(String
tableIdentifier) {
@Override
public synchronized void writeRecord(T row) throws IOException {
addBatch(row);
+ if (isSingle) {
+ size = batch.size();
+ }
boolean valid = (executionOptions.getBatchSize() > 0 && size >=
executionOptions.getBatchSize())
|| batchBytes >= executionOptions.getMaxBatchBytes();
if (valid && !flushing) {
flush();
}
}
+ 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()));
+ }
+ }
+ Object data = jsonFormat ? valueMap : value.toString();
+ LOG.info("appending data object {} with jsonformat {}", data,
jsonFormat);
+ LOG.info("parsed data {}", parsetoMap(data));
+ //appending data object 12345 0
+ //should be {"id":"12345","__DORIS_DELETE_SIGN__":"0"}
+ List<Map<String, String>> mapData =
batchMap.getOrDefault(tableIdentifier, new ArrayList<String>());
+ mapData.add(parsetoMap(data));
Review Comment:
Maybe the data is a map already?
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -198,15 +320,92 @@ private boolean checkFlushException(String
tableIdentifier) {
@Override
public synchronized void writeRecord(T row) throws IOException {
addBatch(row);
+ if (isSingle) {
+ size = batch.size();
Review Comment:
The 'batch' is never used?
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisStreamLoad.java:
##########
@@ -119,6 +121,7 @@ private LoadResponse loadBatch(String db, String tbl,
String value) {
try {
final String loadUrlStr = String.format(LOAD_URL_PATTERN,
hostPort, db, tbl);
+ LOG.info("load batch start, loadURL:{}", loadUrlStr);
Review Comment:
This is a test code?
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -359,7 +558,10 @@ public synchronized void flush() {
flushing = false;
return;
}
+
+ //sample batchmap:
[{"id":"543","__DORIS_DELETE_SIGN__":"0"},{"id":"555","__DORIS_DELETE_SIGN__":"0"}]
Review Comment:
This comment is required?
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -198,15 +320,92 @@ private boolean checkFlushException(String
tableIdentifier) {
@Override
public synchronized void writeRecord(T row) throws IOException {
addBatch(row);
+ if (isSingle) {
+ size = batch.size();
+ }
boolean valid = (executionOptions.getBatchSize() > 0 && size >=
executionOptions.getBatchSize())
|| batchBytes >= executionOptions.getMaxBatchBytes();
if (valid && !flushing) {
flush();
}
}
+ 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()));
+ }
+ }
+ Object data = jsonFormat ? valueMap : value.toString();
+ LOG.info("appending data object {} with jsonformat {}", data,
jsonFormat);
+ LOG.info("parsed data {}", parsetoMap(data));
+ //appending data object 12345 0
Review Comment:
Log printing too often？
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -198,15 +320,92 @@ private boolean checkFlushException(String
tableIdentifier) {
@Override
public synchronized void writeRecord(T row) throws IOException {
addBatch(row);
+ if (isSingle) {
+ size = batch.size();
+ }
boolean valid = (executionOptions.getBatchSize() > 0 && size >=
executionOptions.getBatchSize())
|| batchBytes >= executionOptions.getMaxBatchBytes();
if (valid && !flushing) {
flush();
}
}
+ 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()));
+ }
+ }
+ Object data = jsonFormat ? valueMap : value.toString();
+ LOG.info("appending data object {} with jsonformat {}", data,
jsonFormat);
+ LOG.info("parsed data {}", parsetoMap(data));
+ //appending data object 12345 0
+ //should be {"id":"12345","__DORIS_DELETE_SIGN__":"0"}
+ List<Map<String, String>> mapData =
batchMap.getOrDefault(tableIdentifier, new ArrayList<String>());
+ mapData.add(parsetoMap(data));
+ batchMap.putIfAbsent(tableIdentifier, mapData);
+ } else if (row instanceof String) {
+ batchBytes += ((String)
row).getBytes(StandardCharsets.UTF_8).length;
+ LOG.info("appending row {}", row);
Review Comment:
Log printing too often？
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisStreamLoad.java:
##########
@@ -109,6 +110,7 @@ public void setHostPort(String hostPort) {
}
private LoadResponse loadBatch(String db, String tbl, String value) {
+ LOG.info("load batch start");
Review Comment:
This is a test code?
--
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]
