This is an automated email from the ASF dual-hosted git repository.
amaechler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new a758f52dfc2 fix: keep default values of proto3 fields without presence
in protobuf ingestion (#20485)
a758f52dfc2 is described below
commit a758f52dfc2db46ce61386fe483b0ac88f8f6756
Author: Andreas Maechler <[email protected]>
AuthorDate: Tue Oct 6 14:37:13 2026 -0600
fix: keep default values of proto3 fields without presence in protobuf
ingestion (#20485)
ProtobufConverter built rows from Message.getAllFields(), which omits
singular
fields without presence tracking (proto3 fields not marked optional) when
they
hold their default value. Those columns were ingested as null instead of 0,
"",
false or the first enum value. Add such fields explicitly.
Tests: Ingests proto3 records through Kafka and Kinesis supervisors and
queries
the default-valued fields. Also fixes the ProbufData typo in MoreResources
and
adds the missing timeout on test_protobufDataFormat.
---
.../testing/embedded/indexing/MoreResources.java | 4 +-
.../indexing/StreamIndexDataFormatsTestBase.java | 51 +++++++++++++++++-
.../test/resources/data/protobuf/proto3_event.desc | Bin 0 -> 245 bytes
.../resources/data/protobuf/proto3_event.proto | 32 +++++++++++
.../data/input/protobuf/ProtobufConverter.java | 9 +++-
.../data/input/protobuf/ProtobufConverterTest.java | 59 +++++++++++++++++++++
.../src/test/resources/proto3_test_event.proto | 37 +++++++++++++
7 files changed, 188 insertions(+), 4 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/MoreResources.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/MoreResources.java
index 205271bcfcf..bfb473c8a57 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/MoreResources.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/MoreResources.java
@@ -179,9 +179,11 @@ public class MoreResources
);
}
- public static class ProbufData
+ public static class ProtobufData
{
public static final String WIKI_PROTOBUF_BYTES_DECODER_RESOURCE =
"data/protobuf/wikipedia.desc";
public static final String WIKI_PROTO_MESSAGE_TYPE = "Wikipedia";
+ public static final String PROTO3_EVENT_BYTES_DECODER_RESOURCE =
"data/protobuf/proto3_event.desc";
+ public static final String PROTO3_EVENT_MESSAGE_TYPE = "Proto3Event";
}
}
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexDataFormatsTestBase.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexDataFormatsTestBase.java
index 779a3ebde41..0ce540dc2f3 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexDataFormatsTestBase.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/StreamIndexDataFormatsTestBase.java
@@ -19,6 +19,8 @@
package org.apache.druid.testing.embedded.indexing;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
import org.apache.druid.data.input.InputFormat;
import org.apache.druid.data.input.avro.AvroExtensionsModule;
import org.apache.druid.data.input.avro.AvroStreamInputFormat;
@@ -243,6 +245,7 @@ public abstract class StreamIndexDataFormatsTestBase
extends EmbeddedClusterTest
}
@Test
+ @Timeout(30)
public void test_protobufDataFormat()
{
streamResource.createTopicWithPartitions(dataSource, 3);
@@ -251,8 +254,8 @@ public abstract class StreamIndexDataFormatsTestBase
extends EmbeddedClusterTest
int recordCount = generateStreamAndPublish(dataSource, serializer, false);
FileBasedProtobufBytesDecoder protobufBytesDecoder = new
FileBasedProtobufBytesDecoder(
- MoreResources.ProbufData.WIKI_PROTOBUF_BYTES_DECODER_RESOURCE,
- MoreResources.ProbufData.WIKI_PROTO_MESSAGE_TYPE
+ MoreResources.ProtobufData.WIKI_PROTOBUF_BYTES_DECODER_RESOURCE,
+ MoreResources.ProtobufData.WIKI_PROTO_MESSAGE_TYPE
);
ProtobufInputFormat inputFormat = new ProtobufInputFormat(null,
protobufBytesDecoder);
@@ -291,6 +294,50 @@ public abstract class StreamIndexDataFormatsTestBase
extends EmbeddedClusterTest
stopSupervisor(supervisorSpec);
}
+ @Test
+ @Timeout(30)
+ public void test_protobufProto3DefaultValues()
+ {
+ streamResource.createTopicWithPartitions(dataSource, 3);
+ final FileBasedProtobufBytesDecoder protobufBytesDecoder = new
FileBasedProtobufBytesDecoder(
+ MoreResources.ProtobufData.PROTO3_EVENT_BYTES_DECODER_RESOURCE,
+ MoreResources.ProtobufData.PROTO3_EVENT_MESSAGE_TYPE
+ );
+ final Descriptors.Descriptor descriptor =
protobufBytesDecoder.getDescriptor();
+
+ // The first record leaves every field but the timestamp at its proto3
default, which is not written to the wire
+ final List<byte[]> records = List.of(
+ DynamicMessage.newBuilder(descriptor)
+ .setField(descriptor.findFieldByName("timestamp"),
"2026-01-01T00:00:00Z")
+ .build()
+ .toByteArray(),
+ DynamicMessage.newBuilder(descriptor)
+ .setField(descriptor.findFieldByName("timestamp"),
"2026-01-01T00:00:01Z")
+ .setField(descriptor.findFieldByName("someInt"), 5)
+ .setField(descriptor.findFieldByName("someString"), "x")
+ .setField(descriptor.findFieldByName("someBool"), true)
+ .setField(
+ descriptor.findFieldByName("category"),
+
descriptor.findEnumTypeByName("Category").findValueByName("ONE")
+ )
+ .build()
+ .toByteArray()
+ );
+ streamResource.publishRecordsToTopic(dataSource, records);
+
+ final ProtobufInputFormat inputFormat = new ProtobufInputFormat(null,
protobufBytesDecoder);
+ final SupervisorSpec supervisorSpec = createSupervisor(dataSource,
dataSource, inputFormat);
+ Assertions.assertEquals(dataSource,
cluster.callApi().postSupervisor(supervisorSpec));
+
+ waitForDataAndVerifyIngestedEvents(dataSource, records.size());
+ cluster.callApi().verifySqlQuery(
+ "SELECT someInt, someString IS NULL, someBool, category FROM %s ORDER
BY __time",
+ dataSource,
+ "0,false,false,ZERO\n5,false,true,ONE"
+ );
+ stopSupervisor(supervisorSpec);
+ }
+
@Test
@Timeout(30)
public void test_tsvDataFormat()
diff --git a/embedded-tests/src/test/resources/data/protobuf/proto3_event.desc
b/embedded-tests/src/test/resources/data/protobuf/proto3_event.desc
new file mode 100644
index 00000000000..b373a47c50d
Binary files /dev/null and
b/embedded-tests/src/test/resources/data/protobuf/proto3_event.desc differ
diff --git a/embedded-tests/src/test/resources/data/protobuf/proto3_event.proto
b/embedded-tests/src/test/resources/data/protobuf/proto3_event.proto
new file mode 100644
index 00000000000..581d4a9a593
--- /dev/null
+++ b/embedded-tests/src/test/resources/data/protobuf/proto3_event.proto
@@ -0,0 +1,32 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Regenerate the descriptor from this directory with:
+// protoc --include_imports --descriptor_set_out=proto3_event.desc
proto3_event.proto
+
+syntax = "proto3";
+
+message Proto3Event {
+ enum Category {
+ ZERO = 0;
+ ONE = 1;
+ }
+
+ string timestamp = 1;
+ int32 someInt = 2;
+ string someString = 3;
+ bool someBool = 4;
+ Category category = 5;
+}
diff --git
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/ProtobufConverter.java
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/ProtobufConverter.java
index a032e33ac8a..dfc2923fad2 100644
---
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/ProtobufConverter.java
+++
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/ProtobufConverter.java
@@ -66,10 +66,17 @@ public class ProtobufConverter
return null;
}
final Map<Descriptors.FieldDescriptor, Object> fields = msg.getAllFields();
- final Map<String, Object> converted =
Maps.newHashMapWithExpectedSize(fields.size());
+ final List<Descriptors.FieldDescriptor> declaredFields =
msg.getDescriptorForType().getFields();
+ final Map<String, Object> converted =
Maps.newHashMapWithExpectedSize(Math.max(fields.size(), declaredFields.size()));
for (Map.Entry<Descriptors.FieldDescriptor, Object> field :
fields.entrySet()) {
converted.put(field.getKey().getJsonName(), convertField(field.getKey(),
field.getValue()));
}
+ // getAllFields() omits non-optional proto3 fields set to their default
value, add them back
+ for (final Descriptors.FieldDescriptor field : declaredFields) {
+ if (!field.hasPresence() && !field.isRepeated() &&
!fields.containsKey(field)) {
+ converted.put(field.getJsonName(), convertField(field,
msg.getField(field)));
+ }
+ }
return converted;
}
diff --git
a/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/ProtobufConverterTest.java
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/ProtobufConverterTest.java
new file mode 100644
index 00000000000..12a73edf572
--- /dev/null
+++
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/ProtobufConverterTest.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.data.input.protobuf;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.DynamicMessage;
+import
org.apache.druid.data.input.protobuf.Proto3TestEventWrapper.Proto3TestEvent;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+public class ProtobufConverterTest
+{
+ @Test
+ public void testProto3ImplicitPresenceFieldsKeepDefaultValues() throws
Exception
+ {
+ Assertions.assertEquals(
+ ImmutableMap.of(
+ "someInt", 0,
+ "someString", "",
+ "category", "CATEGORY_ZERO"
+ ),
+ convert(Proto3TestEvent.getDefaultInstance())
+ );
+ }
+
+ @Test
+ public void testProto3NestedMessageKeepsDefaultValues() throws Exception
+ {
+ final Proto3TestEvent event =
+
Proto3TestEvent.newBuilder().setNested(Proto3TestEvent.Nested.getDefaultInstance()).build();
+
+ Assertions.assertEquals(ImmutableMap.of("value", 0),
convert(event).get("nested"));
+ }
+
+ private static Map<String, Object> convert(Proto3TestEvent event) throws
Exception
+ {
+ // Round-trip through bytes like ingestion does.
+ return
ProtobufConverter.convertMessage(DynamicMessage.parseFrom(Proto3TestEvent.getDescriptor(),
event.toByteArray()));
+ }
+}
diff --git
a/extensions-core/protobuf-extensions/src/test/resources/proto3_test_event.proto
b/extensions-core/protobuf-extensions/src/test/resources/proto3_test_event.proto
new file mode 100644
index 00000000000..e96b03d0a9d
--- /dev/null
+++
b/extensions-core/protobuf-extensions/src/test/resources/proto3_test_event.proto
@@ -0,0 +1,37 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+package proto3test;
+option java_package = "org.apache.druid.data.input.protobuf";
+option java_outer_classname = "Proto3TestEventWrapper";
+
+message Proto3TestEvent {
+ enum Category {
+ CATEGORY_ZERO = 0;
+ CATEGORY_ONE = 1;
+ }
+
+ message Nested {
+ int32 value = 1;
+ }
+
+ int32 someInt = 1;
+ string someString = 2;
+ Category category = 3;
+ optional int32 optionalInt = 4;
+ Nested nested = 5;
+ repeated int32 someList = 6;
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
