fix: keep default values of proto3 fields without presence in protobuf ingestion

embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/MoreResources.java

@@ -179,9 +179,11 @@ public static class Supervisor
             );
   }
 
-  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";
   }
 }

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 void test_jsonDataFormat()
   }
 
   @Test
+  @Timeout(30)
   public void test_protobufDataFormat()
   {
     streamResource.createTopicWithPartitions(dataSource, 3);
@@ -251,8 +254,8 @@ public void test_protobufDataFormat()
     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 void test_protobufDataFormatWithSchemaRegistry()
     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()

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;
+}

extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/ProtobufConverter.java

@@ -66,10 +66,17 @@ public static Map<String, Object> convertMessage(Message msg) throws InvalidProt
       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;
   }
 

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()));
+  }
+}

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;
+}