Kafka protobuf consumer support

extensions/kafka/src/main/java/io/deephaven/kafka/ProtobufImpl.java

@@ -14,6 +14,7 @@
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider;
 import io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer;
 import io.deephaven.UncheckedDeephavenException;
+import io.deephaven.api.ColumnName;
 import io.deephaven.engine.table.ColumnDefinition;
 import io.deephaven.engine.table.TableDefinition;
 import io.deephaven.functions.ToBooleanFunction;
@@ -35,6 +36,7 @@
 import io.deephaven.kafka.ingest.KeyOrValueProcessor;
 import io.deephaven.kafka.ingest.MultiFieldChunkAdapter;
 import io.deephaven.kafka.protobuf.ProtobufConsumeOptions;
+import io.deephaven.kafka.protobuf.ProtobufConsumeOptions.FieldPathToColumnName;
 import io.deephaven.protobuf.FieldNumberPath;
 import io.deephaven.protobuf.FieldOptions;
 import io.deephaven.protobuf.FieldPath;
@@ -118,9 +120,12 @@ KeyOrValueIngestData getIngestData(KeyOrValue keyOrValue, SchemaRegistryClient s
             final List<FieldCopier> fieldCopiers = new ArrayList<>(functions.functions().size());
             final KeyOrValueIngestData data = new KeyOrValueIngestData();
             data.fieldPathToColumnName = new LinkedHashMap<>();
-            final Function<FieldPath, String> pathToColumnName = specs.pathToColumnName();
+            final FieldPathToColumnName fieldPathToColumnName = specs.pathToColumnName();
+            final Map<FieldPath, Integer> counts = new HashMap<>();
             for (ProtobufFunction f : functions.functions()) {
-                add(pathToColumnName.apply(f.path()), f.function(), data, columnDefinitionsOut, fieldCopiers);
+                final int ix = counts.compute(f.path(), (fieldPath, ixCount) -> ixCount == null ? 0 : ixCount + 1);
+                final ColumnName columnName = fieldPathToColumnName.columnName(f.path(), ix);
+                add(columnName, f.function(), data, columnDefinitionsOut, fieldCopiers);
             }
             // we don't have enough info at this time to create KeyOrValueProcessorImpl
             // data.extra = new KeyOrValueProcessorImpl(MultiFieldChunkAdapter.chunkOffsets(null, null), fieldCopiers,
@@ -130,13 +135,13 @@ KeyOrValueIngestData getIngestData(KeyOrValue keyOrValue, SchemaRegistryClient s
         }
 
         private void add(
-                String columnName,
+                ColumnName columnName,
                 TypedFunction<Message> function,
                 KeyOrValueIngestData data,
                 List<ColumnDefinition<?>> columnDefinitionsOut,
                 List<FieldCopier> fieldCopiersOut) {
-            data.fieldPathToColumnName.put(columnName, columnName);
-            columnDefinitionsOut.add(ColumnDefinition.of(columnName, function.returnType()));
+            data.fieldPathToColumnName.put(columnName.name(), columnName.name());
+            columnDefinitionsOut.add(ColumnDefinition.of(columnName.name(), function.returnType()));
             fieldCopiersOut.add(FieldCopierAdapter.of(PROTOBUF_MESSAGE_OBJ.map(ToChunkTypeTransform.of(function))));
         }
 

extensions/kafka/src/main/java/io/deephaven/kafka/protobuf/ProtobufConsumeOptions.java

@@ -5,6 +5,7 @@
 
 import com.google.protobuf.Descriptors.Descriptor;
 import io.deephaven.annotations.BuildableStyle;
+import io.deephaven.api.ColumnName;
 import io.deephaven.kafka.KafkaTools.Consume;
 import io.deephaven.protobuf.FieldPath;
 import io.deephaven.protobuf.ProtobufDescriptorParserOptions;
@@ -31,6 +32,20 @@
 @BuildableStyle
 public abstract class ProtobufConsumeOptions {
 
+    @FunctionalInterface
+    public interface FieldPathToColumnName {
+        /**
+         * Creates a unique column name from {@code fieldPath} and {@code indexOccurrence}. Implementations will need to
+         * take notice when {@code indexOccurrence > 0}, as that means a column name for {@code fieldPath} has already
+         * been generated {@code indexOccurrence + 1} times.
+         * 
+         * @param fieldPath the field path
+         * @param indexOccurrence the number of times a column name for fieldPath has already been generated
+         * @return the column name
+         */
+        ColumnName columnName(FieldPath fieldPath, int indexOccurrence);
+    }
+
     /**
      * The builder.
      *
@@ -41,13 +56,15 @@ public static Builder builder() {
     }
 
     /**
-     * Joins the name paths with underscores. Equivalent to {@code String.join("_", path.namePath())}.
+     * Joins the name paths with underscores, appending {@code indexOccurrence + 1} if {@code indexOccurrence != 0}.
      *
      * @param path the path
+     * @param indexOccurrence the number of times this field path has been used
      * @return the underscore joined path names
      */
-    public static String joinNamePathWithUnderscore(FieldPath path) {
-        return String.join("_", path.namePath());
+    public static ColumnName joinNamePathWithUnderscore(FieldPath path, int indexOccurrence) {
+        final String simple = String.join("_", path.namePath());
+        return ColumnName.of(indexOccurrence == 0 ? simple : simple + "_" + (indexOccurrence + 1));
     }
 
     /**
@@ -94,12 +111,12 @@ public ProtobufDescriptorParserOptions parserOptions() {
 
     /**
      * The function to turn field paths into column names. By default, is the function
-     * {@link #joinNamePathWithUnderscore(FieldPath)}.
+     * {@link #joinNamePathWithUnderscore(FieldPath, int)}}.
      *
      * @return the function to create column names
      */
     @Default
-    public Function<FieldPath, String> pathToColumnName() {
+    public FieldPathToColumnName pathToColumnName() {
         return ProtobufConsumeOptions::joinNamePathWithUnderscore;
     }
 
@@ -113,7 +130,7 @@ public interface Builder {
 
         Builder parserOptions(ProtobufDescriptorParserOptions options);
 
-        Builder pathToColumnName(Function<FieldPath, String> pathToColumnName);
+        Builder pathToColumnName(FieldPathToColumnName pathToColumnName);
 
         ProtobufConsumeOptions build();
     }

extensions/protobuf/src/main/java/io/deephaven/protobuf/Builtin.java

@@ -26,7 +26,6 @@
 import io.deephaven.functions.ToIntFunction;
 import io.deephaven.functions.ToLongFunction;
 import io.deephaven.functions.ToObjectFunction;
-import io.deephaven.functions.TypedFunction;
 import io.deephaven.protobuf.FieldOptions.BytesBehavior;
 import io.deephaven.qst.type.CustomType;
 import io.deephaven.qst.type.GenericType;
@@ -44,7 +43,7 @@ class Builtin {
     // Due to https://github.com/confluentinc/schema-registry/issues/2708, the parsers need to be built in such a way
     // that they work with DynamicMessages.
 
-    static List<SingleValuedMessageParser> parsers() {
+    static List<MessageParserSingle> parsers() {
         // Update javadoc in io.deephaven.protobuf.SingleValuedMessageParser.builtin when editing
         return List.of(
                 TimestampParser.of(),
@@ -62,7 +61,7 @@ static List<SingleValuedMessageParser> parsers() {
                 customParser(FieldMask.class));
     }
 
-    static <T extends Message> SingleValuedMessageParser customParser(Class<T> clazz) {
+    static <T extends Message> MessageParserSingle customParser(Class<T> clazz) {
         try {
             final Method method = clazz.getDeclaredMethod("getDescriptor");
             final Descriptor descriptor = (Descriptor) method.invoke(null);
@@ -72,14 +71,14 @@ static <T extends Message> SingleValuedMessageParser customParser(Class<T> clazz
         }
     }
 
-    private enum TimestampParser implements SingleValuedMessageParser {
+    private enum TimestampParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToObjectFunction<Message, Instant> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, Timestamp>identity(Type.ofCustom(Timestamp.class))
                 .mapToObj(TimestampParser::parseCanonical, Type.instantType());
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -104,15 +103,15 @@ public ToObjectFunction<Message, Instant> messageParser(Descriptor descriptor,
         }
     }
 
-    private enum DurationParser implements SingleValuedMessageParser {
+    private enum DurationParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToObjectFunction<Message, Duration> CANONICAL_FUNCTION =
                 ToObjectFunction.<Message, com.google.protobuf.Duration>identity(
                         Type.ofCustom(com.google.protobuf.Duration.class))
                         .mapToObj(DurationParser::parseCanonical, Type.ofCustom(Duration.class));
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -137,14 +136,14 @@ public ToObjectFunction<Message, Duration> messageParser(Descriptor descriptor,
         }
     }
 
-    private enum BoolValueParser implements SingleValuedMessageParser {
+    private enum BoolValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToBooleanFunction<Message> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, BoolValue>identity(Type.ofCustom(BoolValue.class))
                 .mapToBoolean(BoolValue::getValue);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -164,14 +163,14 @@ public ToBooleanFunction<Message> messageParser(Descriptor descriptor, ProtobufD
         }
     }
 
-    private enum Int32ValueParser implements SingleValuedMessageParser {
+    private enum Int32ValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToIntFunction<Message> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, Int32Value>identity(Type.ofCustom(Int32Value.class))
                 .mapToInt(Int32Value::getValue);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -191,14 +190,14 @@ public ToIntFunction<Message> messageParser(Descriptor descriptor, ProtobufDescr
         }
     }
 
-    private enum UInt32ValueParser implements SingleValuedMessageParser {
+    private enum UInt32ValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToIntFunction<Message> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, UInt32Value>identity(Type.ofCustom(UInt32Value.class))
                 .mapToInt(UInt32Value::getValue);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -218,14 +217,14 @@ public ToIntFunction<Message> messageParser(Descriptor descriptor, ProtobufDescr
         }
     }
 
-    private enum Int64ValueParser implements SingleValuedMessageParser {
+    private enum Int64ValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToLongFunction<Message> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, Int64Value>identity(Type.ofCustom(Int64Value.class))
                 .mapToLong(Int64Value::getValue);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -245,14 +244,14 @@ public ToLongFunction<Message> messageParser(Descriptor descriptor, ProtobufDesc
         }
     }
 
-    private enum UInt64ValueParser implements SingleValuedMessageParser {
+    private enum UInt64ValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToLongFunction<Message> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, UInt64Value>identity(Type.ofCustom(UInt64Value.class))
                 .mapToLong(UInt64Value::getValue);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -272,14 +271,14 @@ public ToLongFunction<Message> messageParser(Descriptor descriptor, ProtobufDesc
         }
     }
 
-    private enum FloatValueParser implements SingleValuedMessageParser {
+    private enum FloatValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToFloatFunction<Message> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, FloatValue>identity(Type.ofCustom(FloatValue.class))
                 .mapToFloat(FloatValue::getValue);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -299,14 +298,14 @@ public ToFloatFunction<Message> messageParser(Descriptor descriptor, ProtobufDes
         }
     }
 
-    private enum DoubleValueParser implements SingleValuedMessageParser {
+    private enum DoubleValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToDoubleFunction<Message> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, DoubleValue>identity(Type.ofCustom(DoubleValue.class))
                 .mapToDouble(DoubleValue::getValue);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -326,14 +325,14 @@ public ToDoubleFunction<Message> messageParser(Descriptor descriptor, ProtobufDe
         }
     }
 
-    private enum StringValueParser implements SingleValuedMessageParser {
+    private enum StringValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToObjectFunction<Message, String> CANONICAL_FUNCTION = ToObjectFunction
                 .<Message, StringValue>identity(Type.ofCustom(StringValue.class))
                 .mapToObj(StringValue::getValue, Type.stringType());
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -354,7 +353,7 @@ public ToObjectFunction<Message, String> messageParser(Descriptor descriptor,
         }
     }
 
-    private enum BytesValueParser implements SingleValuedMessageParser {
+    private enum BytesValueParser implements MessageParserSingle {
         INSTANCE;
 
         private static final ToObjectFunction<Message, ByteString> CANONICAL_FUNCTION_BYTESTRING = ToObjectFunction
@@ -365,7 +364,7 @@ private enum BytesValueParser implements SingleValuedMessageParser {
         private static final ToObjectFunction<Message, byte[]> CANONICAL_FUNCTION_BYTEARRAY =
                 CANONICAL_FUNCTION_BYTESTRING.mapToObj(BYTESTRING_TO_BYTES);
 
-        public static SingleValuedMessageParser of() {
+        public static MessageParserSingle of() {
             return INSTANCE;
         }
 
@@ -393,7 +392,7 @@ public Descriptor canonicalDescriptor() {
         }
     }
 
-    private static class GenericSVMP<T extends Message> implements SingleValuedMessageParser {
+    private static class GenericSVMP<T extends Message> implements MessageParserSingle {
         private final GenericType<T> type;
         private final Descriptor descriptor;