EMsnap commented on code in PR #6358:
URL: https://github.com/apache/inlong/pull/6358#discussion_r1011161522
##########
inlong-sort/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java:
##########
@@ -277,6 +284,53 @@ private void addBatch(T row) throws IOException {
}
}
+ private void handleColumnsChange(String tableIdentifier, JsonNode
rootNode, JsonNode physicalData) {
+ String columns = parseColumns(rootNode, physicalData);
+ String oldColumns = columnsMap.get(tableIdentifier);
+ if (columns == null && oldColumns != null || (columns != null &&
!columns.equals(oldColumns))) {
+ flushSingleTable(tableIdentifier, batchMap.get(tableIdentifier));
+ if (!errorTables.contains(tableIdentifier)) {
+ columnsMap.put(tableIdentifier, columns);
+ } else {
+ batchMap.remove(tableIdentifier);
+ columnsMap.remove(tableIdentifier);
+ errorTables.remove(tableIdentifier);
+ }
+ }
+ }
+
+ private String parseColumns(JsonNode rootNode, JsonNode physicalData) {
+ // Add column key when fieldNames is not empty
+ Iterator<String> fieldNames = null;
+ try {
+ RowType rowType = jsonDynamicSchemaFormat.extractSchema(rootNode);
+ if (rowType != null) {
+ fieldNames = rowType.getFieldNames().listIterator();
+ }
+ } catch (IllegalArgumentException e) {
+ LOG.warn("extract schema failed", e);
+ // Extract schema from physicalData
+ JsonNode first = physicalData.isArray() ? physicalData.get(0) :
physicalData;
+ // Add column key when fieldNames is not empty
+ fieldNames = first.fieldNames();
+ }
+ if (fieldNames != null && fieldNames.hasNext()) {
+ StringBuilder sb = new StringBuilder();
Review Comment:
a method for this logic to construct doris columns would be better
--
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]
