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 a6239e016c [INLONG-12141][SDK] PbSourceData returns protobuf default
values instead of null for unset fields (#12142)
a6239e016c is described below
commit a6239e016c2efe8518188bc24205c914ae3555b7
Author: ChunLiang Lu <[email protected]>
AuthorDate: Thu Jun 25 10:36:18 2026 +0800
[INLONG-12141][SDK] PbSourceData returns protobuf default values instead of
null for unset fields (#12142)
* [INLONG-12141][SDK] PbSourceData returns protobuf default values instead
of null for unset fields, causing incorrect data written to Iceberg/Parquet
sinks
* [INLONG-12141][SDK] PbSourceData returns protobuf default values instead
of null for unset fields, causing incorrect data written to Iceberg/Parquet
sinks
---
.../inlong/sdk/transform/decode/PbSourceData.java | 31 +++++++++++++++++-----
.../sdk/transform/decode/PbSourceDecoder.java | 2 +-
.../sdk/transform/process/TransformProcessor.java | 2 +-
.../temporal/TestFromUnixTimeFunction.java | 9 +++++++
.../process/processor/TestPb2RowDataProcessor.java | 2 +-
5 files changed, 36 insertions(+), 10 deletions(-)
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 f5d3569c2c..7f6fbcdf85 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
@@ -256,6 +256,10 @@ public class PbSourceData extends AbstractSourceData {
GenericRowData result = new
GenericRowData(messageType.getFields().size());
int index = 0;
for (FieldDescriptor fieldDesc : messageType.getFields()) {
+ if (!fieldDesc.isRepeated() && !msgObj.hasField(fieldDesc)) {
+ result.setField(index++, null);
+ continue;
+ }
Object fieldValue = msgObj.getField(fieldDesc);
if (fieldValue == null) {
result.setField(index++, null);
@@ -301,10 +305,12 @@ public class PbSourceData extends AbstractSourceData {
continue;
}
DynamicMessage subnodeValue = (DynamicMessage) value;
- Object keyValue = buildFieldValue(keyField,
subnodeValue.getField(keyField));
- Object valueValue = buildFieldValue(valueField,
subnodeValue.getField(valueField));
- if (keyValue != null && valueValue != null) {
- result.put(keyValue, valueValue);
+ if (subnodeValue.hasField(keyField) &&
subnodeValue.hasField(valueField)) {
+ Object keyValue = buildFieldValue(keyField,
subnodeValue.getField(keyField));
+ Object valueValue = buildFieldValue(valueField,
subnodeValue.getField(valueField));
+ if (keyValue != null && valueValue != null) {
+ result.put(keyValue, valueValue);
+ }
}
}
return new GenericMapData(result);
@@ -562,9 +568,11 @@ public class PbSourceData extends AbstractSourceData {
continue;
}
DynamicMessage msg = (DynamicMessage) value;
- Object keyValue = msg.getField(node.getMapKeyDesc());
- Object valueValue = msg.getField(node.getMapValueDesc());
- mapCache.put(keyValue, valueValue);
+ if (msg.hasField(node.getMapKeyDesc()) &&
msg.hasField(node.getMapValueDesc())) {
+ Object keyValue = msg.getField(node.getMapKeyDesc());
+ Object valueValue = msg.getField(node.getMapValueDesc());
+ mapCache.put(keyValue, valueValue);
+ }
}
return mapCache;
}
@@ -578,6 +586,9 @@ public class PbSourceData extends AbstractSourceData {
DynamicMessage current = root;
for (int i = 0; i < childNodes.size(); i++) {
PbNode node = childNodes.get(i);
+ if (!node.getFieldDesc().isRepeated() &&
!current.hasField(node.getFieldDesc())) {
+ break;
+ }
Object nodeValue = current.getField(node.getFieldDesc());
if (nodeValue == null) {
// error data
@@ -815,6 +826,9 @@ public class PbSourceData extends AbstractSourceData {
continue;
}
DynamicMessage entryMsg = (DynamicMessage) entry;
+ if (!entryMsg.hasField(mapKeyDesc) ||
!entryMsg.hasField(mapValueDesc)) {
+ continue;
+ }
Object keyVal = entryMsg.getField(mapKeyDesc);
if (keyVal == null || !Objects.equals(node.getMapKey(),
keyVal)) {
continue;
@@ -873,6 +887,9 @@ public class PbSourceData extends AbstractSourceData {
for (int i = 0; i < count; i++) {
Object entry = builder.getRepeatedField(fd, i);
if (entry instanceof DynamicMessage) {
+ if (!((DynamicMessage) entry).hasField(mapKeyDesc)) {
+ continue;
+ }
Object keyVal = ((DynamicMessage) entry).getField(mapKeyDesc);
if (Objects.equals(targetKey, keyVal)) {
removed = true;
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceDecoder.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceDecoder.java
index 46ce0215f1..146063201f 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceDecoder.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/PbSourceDecoder.java
@@ -107,7 +107,7 @@ public class PbSourceDecoder extends SourceDecoder<String> {
try {
// decode
DynamicMessage.Builder builder =
DynamicMessage.newBuilder(rootDesc);
- DynamicMessage root = builder.mergeFrom(srcBytes).build();
+ DynamicMessage root = builder.mergeFrom(srcBytes).buildPartial();
// child
List<DynamicMessage> childRoot = null;
if (this.childNodes != null && this.childNodes.size() > 0) {
diff --git
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/TransformProcessor.java
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/TransformProcessor.java
index cc70ee3f95..f0e39c04bd 100644
---
a/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/TransformProcessor.java
+++
b/inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/process/TransformProcessor.java
@@ -230,7 +230,7 @@ public class TransformProcessor<I, O> {
try {
Object fieldValue = parser.parse(sourceData, i, context);
if (fieldValue == null) {
- sinkData.addField(fieldName, "");
+ sinkData.addField(fieldName, null);
} else {
sinkData.addField(fieldName, fieldValue);
}
diff --git
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/function/temporal/TestFromUnixTimeFunction.java
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/function/temporal/TestFromUnixTimeFunction.java
index 9d8c24d7b4..0be60f8338 100644
---
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/function/temporal/TestFromUnixTimeFunction.java
+++
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/function/temporal/TestFromUnixTimeFunction.java
@@ -91,5 +91,14 @@ public class TestFromUnixTimeFunction extends
AbstractFunctionTemporalTestBase {
List<String> output6 =
processor6.transform("can|apple|cloud|1753353737123|1|3", new HashMap<>());
Assert.assertEquals(1, output6.size());
Assert.assertEquals(output6.get(0), "result=2025-07-24 18:00:00");
+
+ String transformSql7 = "select from_unix_time(now()/1000,'yyyyMMddHH')
as partitionTime from source";
+ TransformConfig config7 = new TransformConfig(transformSql7);
+ TransformProcessor<String, String> processor7 = TransformProcessor
+ .create(config7,
SourceDecoderFactory.createCsvDecoder(csvSource),
+ SinkEncoderFactory.createKvEncoder(kvSink));
+ // case7: from_unix_time(now()/1000,'yyyyMMddHH')
+ List<String> output7 =
processor7.transform("can|apple|cloud|1753353737123|1|3", new HashMap<>());
+ Assert.assertEquals(1, output7.size());
}
}
diff --git
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/TestPb2RowDataProcessor.java
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/TestPb2RowDataProcessor.java
index 88d023bf38..495172d627 100644
---
a/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/TestPb2RowDataProcessor.java
+++
b/inlong-sdk/transform-sdk/src/test/java/org/apache/inlong/sdk/transform/process/processor/TestPb2RowDataProcessor.java
@@ -247,7 +247,7 @@ public class TestPb2RowDataProcessor extends
AbstractProcessorTestBase {
Assert.assertEquals(output.get(0).getString(1).toString(), "1");
Assert.assertEquals(output.get(0).getString(2).toString(),
"1713243918000");
- Assert.assertEquals(((GenericRowData) output.get(0).getRow(3,
3)).getBinary(0).length, 0);
+ Assert.assertEquals(((GenericRowData) output.get(0).getRow(3,
3)).getBinary(0), null);
Assert.assertEquals(((GenericRowData) output.get(0).getRow(3,
3)).getLong(1), 1713243918000L);
Assert.assertEquals(((GenericRowData) output.get(0).getRow(3,
3)).getMap(2).size(), 0);
