This is an automated email from the ASF dual-hosted git repository.
luchunliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 066cb0b9b4 [INLONG-12111][SDK] ProtoBuf decoding supports parsing map
nodes (#12112)
066cb0b9b4 is described below
commit 066cb0b9b49945eca6376e62e91c0f773f2ce6ca
Author: ChunLiang Lu <[email protected]>
AuthorDate: Wed Apr 15 11:09:48 2026 +0800
[INLONG-12111][SDK] ProtoBuf decoding supports parsing map nodes (#12112)
---
.../apache/inlong/sdk/transform/decode/PbNode.java | 62 ++++++++++++++++++----
.../inlong/sdk/transform/decode/PbSourceData.java | 52 ++++++++++--------
2 files changed, 82 insertions(+), 32 deletions(-)
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbNode.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbNode.java
index 63fa6106d9..4fe166832b 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbNode.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbNode.java
@@ -37,30 +37,52 @@ public class PbNode {
private String name;
private FieldDescriptor fieldDesc;
private Descriptors.Descriptor messageType;
+ private boolean isLastNode = false;
private boolean isArray = false;
private int arrayIndex = -1;
- private boolean isLastNode = false;
+ private boolean isMap = false;
+ private String mapKey = "";
+ private FieldDescriptor mapKeyDesc;
+ private FieldDescriptor mapValueDesc;
public PbNode(Descriptors.Descriptor messageDesc, String nodeString,
boolean isLastNode) {
int beginIndex = nodeString.indexOf('(');
if (beginIndex < 0) {
this.name = nodeString;
- this.fieldDesc = messageDesc.findFieldByName(name);
- if (this.fieldDesc.getJavaType() == JavaType.MESSAGE) {
- this.messageType = this.fieldDesc.getMessageType();
+ if (isMapDescriptor(messageDesc)) {
+ FieldDescriptor valueFieldDesc =
messageDesc.getFields().get(1);
+ Descriptors.Descriptor valueTypeDesc =
valueFieldDesc.getMessageType();
+ this.fieldDesc = valueTypeDesc.findFieldByName(name);
+ if (this.fieldDesc.getJavaType() == JavaType.MESSAGE) {
+ this.messageType = this.fieldDesc.getMessageType();
+ }
+ } else {
+ this.fieldDesc = messageDesc.findFieldByName(name);
+ if (this.fieldDesc.getJavaType() == JavaType.MESSAGE) {
+ this.messageType = this.fieldDesc.getMessageType();
+ }
}
} else {
this.name = StringUtils.trim(nodeString.substring(0, beginIndex));
this.fieldDesc = messageDesc.findFieldByName(name);
if (this.fieldDesc.getJavaType() == JavaType.MESSAGE) {
this.messageType = this.fieldDesc.getMessageType();
- }
- int endIndex = nodeString.lastIndexOf(')');
- if (endIndex >= 0) {
- this.isArray = true;
- this.arrayIndex =
NumberUtils.toInt(nodeString.substring(beginIndex + 1, endIndex), -1);
- if (this.arrayIndex < 0) {
- this.arrayIndex = 0;
+ int endIndex = nodeString.lastIndexOf(')');
+ if (isMapDescriptor(messageType)) {
+ this.isMap = true;
+ if (endIndex >= 0) {
+ this.mapKey = nodeString.substring(beginIndex + 1,
endIndex);
+ this.mapKeyDesc = messageType.getFields().get(0);
+ this.mapValueDesc = messageType.getFields().get(1);
+ }
+ } else {
+ this.isArray = true;
+ if (endIndex >= 0) {
+ this.arrayIndex =
NumberUtils.toInt(nodeString.substring(beginIndex + 1, endIndex), -1);
+ if (this.arrayIndex < 0) {
+ this.arrayIndex = 0;
+ }
+ }
}
}
}
@@ -92,4 +114,22 @@ public class PbNode {
}
return nodes;
}
+
+ public static boolean isMapDescriptor(Descriptors.Descriptor descriptor) {
+ if (descriptor.getOptions().getMapEntry()) {
+ return true;
+ }
+
+ if (descriptor.getFields().size() == 2) {
+ Descriptors.FieldDescriptor keyField =
descriptor.findFieldByNumber(1);
+ Descriptors.FieldDescriptor valueField =
descriptor.findFieldByNumber(2);
+
+ if (keyField != null && valueField != null &&
+ "key".equals(keyField.getName()) &&
+ "value".equals(valueField.getName())) {
+ return true;
+ }
+ }
+ return false;
+ }
}
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceData.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceData.java
index d4bea9906e..dbeeade6f3 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceData.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceData.java
@@ -184,6 +184,7 @@ public class PbSourceData extends AbstractSourceData {
* @param root
* @return
*/
+ @SuppressWarnings("rawtypes")
private String getNodeValue(List<PbNode> childNodes, DynamicMessage root) {
String fieldValue = "";
DynamicMessage current = root;
@@ -212,33 +213,42 @@ public class PbSourceData extends AbstractSourceData {
fieldValue = String.valueOf(nodeValue);
break;
case MESSAGE:
- fieldValue = String.valueOf(nodeValue);
+ if (node.isArray()) {
+ fieldValue = String.valueOf(((List)
nodeValue).get(node.getArrayIndex()));
+ } else if (node.isMap()) {
+ List<DynamicMessage> nodeValueList =
(List<DynamicMessage>) nodeValue;
+ for (DynamicMessage subnodeValue : nodeValueList) {
+ String keyValue =
String.valueOf(subnodeValue.getField(node.getMapKeyDesc()));
+ if (StringUtils.equals(keyValue,
node.getMapKey())) {
+ fieldValue =
String.valueOf(subnodeValue.getField(node.getMapValueDesc()));
+ break;
+ }
+ }
+ } else {
+ fieldValue = String.valueOf(nodeValue);
+ }
break;
}
break;
}
- if (!node.isArray()) {
- if (!(nodeValue instanceof DynamicMessage)) {
- // error data
- break;
- }
- current = (DynamicMessage) nodeValue;
- } else {
- if (!(nodeValue instanceof List)) {
- // error data
- break;
- }
- List<?> nodeList = (List<?>) nodeValue;
- if (node.getArrayIndex() >= nodeList.size()) {
- // error data
- break;
+ if (node.isArray()) {
+ current = (DynamicMessage) ((List)
nodeValue).get(node.getArrayIndex());
+ } else if (node.isMap()) {
+ List<DynamicMessage> nodeValueList = (List<DynamicMessage>)
nodeValue;
+ DynamicMessage newCurrent = null;
+ for (DynamicMessage subnodeValue : nodeValueList) {
+ String keyValue =
String.valueOf(subnodeValue.getField(node.getMapKeyDesc()));
+ if (StringUtils.equals(keyValue, node.getMapKey())) {
+ newCurrent = (DynamicMessage)
subnodeValue.getField(node.getMapValueDesc());
+ break;
+ }
}
- Object nodeElement = nodeList.get(node.getArrayIndex());
- if (!(nodeElement instanceof DynamicMessage)) {
- // error data
- break;
+ if (newCurrent == null) {
+ return fieldValue;
}
- current = (DynamicMessage) nodeElement;
+ current = newCurrent;
+ } else {
+ current = (DynamicMessage) nodeValue;
}
}
return fieldValue;
