Skip to content

[INLONG-12117][SDK] Support concat_struct/extract_struct/extract_binary function #12118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
luchunliang merged 7 commits into from
May 18, 2026
Merged

[INLONG-12117][SDK] Support concat_struct/extract_struct/extract_binary function #12118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
db2889b
[INLONG-12117][SDK] Support concat_struct/extract_struct/extract_bina…
luchunliang May 14, 2026
fe576ab
fix AI comments
luchunliang May 14, 2026
39af3c7
fix comments
luchunliang May 14, 2026
b60942b
For enum types, return the integer index.
luchunliang May 15, 2026
399fe9d
Allow extract_struct to accept the results of extract_binary as its p…
luchunliang May 15, 2026
c2a6316
Add support for array nodes in the extract_struct function.
luchunliang May 15, 2026
6adee87
New function concat_struct(node1, node2, ...) — picks values from mul…
luchunliang May 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ...ransform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/AbstractSourceData.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,11 +29,11 @@ public abstract class AbstractSourceData implements SourceData {

protected Context context;

protected boolean isContextField(String fieldName) {
public boolean isContextField(String fieldName) {
return fieldName.startsWith(CTX_KEY);
}

protected String getContextField(String fieldName) {
public String getContextField(String fieldName) {
if (context == null) {
return "";
}
Expand Down
196 changes: 149 additions & 47 deletions inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
import lombok.Data;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
Expand All @@ -34,63 +36,118 @@
@Data
public class PbNode {

public static final Logger LOG = LoggerFactory.getLogger(PbNode.class);

private String name;
private FieldDescriptor fieldDesc;
private Descriptors.Descriptor messageType;
private boolean isLastNode = false;
private boolean isArray = false;
private int arrayIndex = -1;
private boolean isMap = false;
// primitive
private boolean isPrimitiveType = false;
// array
private boolean isArrayType = false;
private boolean hasArrayIndex = false;
private Integer arrayIndex;
// struct
private boolean isStructType = false;
// map
private boolean isMapType = false;
private String mapKey = "";
private boolean hasMapKey = false;
private Object mapKey;
private FieldDescriptor mapKeyDesc;
private FieldDescriptor mapValueDesc;
// parent path
private String parentPath;
private String currentPath;
private String currentIndexPath;

public PbNode(Descriptors.Descriptor messageDesc, String nodeString, boolean isLastNode) {
int beginIndex = nodeString.indexOf('(');
if (beginIndex < 0) {
this.name = nodeString;
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();
public PbNode(Descriptors.Descriptor parentDesc, String parentPath, String nodeString, boolean isLastNode) {
try {
if (parentDesc == null) {
return;
}
this.isLastNode = isLastNode;
// parse name & index
int beginIndex = nodeString.indexOf('(');
String indexString = null;
if (beginIndex < 0) {
this.name = nodeString;
} else {
this.name = StringUtils.trim(nodeString.substring(0, beginIndex));
int endIndex = nodeString.lastIndexOf(')');
if (endIndex >= 0) {
indexString = nodeString.substring(beginIndex + 1, endIndex);
}
}
// cache path key
this.parentPath = parentPath;
if (this.parentPath == null) {
this.currentPath = this.name;
this.currentIndexPath = nodeString;
} else {
this.fieldDesc = messageDesc.findFieldByName(name);
if (this.fieldDesc.getJavaType() == JavaType.MESSAGE) {
this.messageType = this.fieldDesc.getMessageType();
if (isMapDescriptor(messageType)) {
this.isMapType = true;
}
this.currentPath = this.parentPath + "." + this.name;
this.currentIndexPath = this.parentPath + "." + nodeString;
}
// field desc
this.fieldDesc = parentDesc.findFieldByName(name);
if (this.fieldDesc == null) {
return;
}
// map
if (this.fieldDesc.getJavaType() == JavaType.MESSAGE
&& isMapDescriptor(this.fieldDesc.getMessageType())) {
this.isMapType = true;
this.mapKeyDesc = this.fieldDesc.getMessageType().getFields().get(0);
this.mapValueDesc = this.fieldDesc.getMessageType().getFields().get(1);
if (indexString != null) {
this.hasMapKey = true;
this.mapKey = parseMapKey(indexString, mapKeyDesc);
}
return;
}
} 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 (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;
}
}
// array
if (this.fieldDesc.isRepeated()) {
this.isArrayType = true;
this.arrayIndex = NumberUtils.toInt(indexString, -1);
if (arrayIndex >= 0) {
this.hasArrayIndex = true;
}
return;
}
// struct
if (this.fieldDesc.getJavaType() == JavaType.MESSAGE) {
this.isStructType = true;
return;
}
// primitive
this.isPrimitiveType = true;
} catch (RuntimeException t) {
LOG.error("Fail to PbNode,error:{},fullName:{},nodePath:{},isLastNode:{}",
t.getMessage(), parentDesc.getName(), nodeString, isLastNode, t);
throw t;
}
}

private static Object parseMapKey(String indexString, FieldDescriptor mapKeyDesc) {
switch (mapKeyDesc.getJavaType()) {
case STRING:
return indexString;
case INT:
return NumberUtils.toInt(indexString, 0);
case LONG:
return NumberUtils.toLong(indexString, 0);
case FLOAT:
return NumberUtils.toFloat(indexString, 0);
case DOUBLE:
return NumberUtils.toDouble(indexString, 0);
case BOOLEAN:
return Boolean.TRUE.toString().equals(indexString);
case ENUM:
return mapKeyDesc.getEnumType().findValueByName(indexString);
case BYTE_STRING:
case MESSAGE:
default:
return indexString;
}
this.isLastNode = isLastNode;
}

/**
Expand All @@ -105,16 +162,61 @@ public static List<PbNode> parseNodePath(Descriptors.Descriptor rootDesc, String
}
List<PbNode> nodes = new ArrayList<>();
String[] nodeStrings = nodePath.split("\\.");
int lastIndex = nodeStrings.length - 1;
final int lastIndex = nodeStrings.length - 1;
String parentPath = null;
StringBuilder currentPathBuilder = new StringBuilder();
Descriptors.Descriptor current = rootDesc;
for (int i = 0; i <= lastIndex; i++) {
if (current == null) {
return null;
}
// pbNode
String nodeString = nodeStrings[i];
PbNode pbNode = new PbNode(current, nodeString, (i == lastIndex));
current = pbNode.getMessageType();
nodes.add(pbNode);
PbNode pbNode = new PbNode(current, parentPath, nodeString, (i == lastIndex));
if (parentPath == null) {
currentPathBuilder.append(nodeString);
} else {
currentPathBuilder.append(".").append(nodeString);
}
parentPath = currentPathBuilder.toString();
if (pbNode.getFieldDesc() == null) {
return null;
}
// primitive
if (pbNode.isPrimitiveType()) {
current = null;
nodes.add(pbNode);
continue;
} else if (pbNode.isArrayType()) {
// array
if (pbNode.getFieldDesc().getJavaType() == JavaType.MESSAGE) {
current = pbNode.getFieldDesc().getMessageType();
} else {
current = null;
}
nodes.add(pbNode);
continue;
} else if (pbNode.isMapType()) {
// map
if (pbNode.isHasMapKey()) {
if (pbNode.getMapValueDesc().getJavaType() == JavaType.MESSAGE) {
current = pbNode.getMapValueDesc().getMessageType();
} else {
current = null;
}
} else {
current = null;
}
nodes.add(pbNode);
continue;
} else if (pbNode.isStructType()) {
// struct
current = pbNode.getFieldDesc().getMessageType();
nodes.add(pbNode);
continue;
} else {
return null;
}
}
return nodes;
}
Expand Down
Loading
Loading