[BEAM-7274] Add DynamicMessage Schema support

Add DynamicMessage schema support. This is different from
generated classes as it uses the proto descriptors. It uses
the ProtoDomain as an index for searching embedded messages.

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/OneOfType.java

@@ -115,8 +115,13 @@ public FieldType getBaseType() {
  }
 
  /** Create a {@link Value} specifying which field to set and the value to set. */
- public <T> Value createValue(String caseType, T value) {
- return createValue(getCaseEnumType().valueOf(caseType), value);
+ public <T> Value createValue(String caseValue, T value) {
+ return createValue(getCaseEnumType().valueOf(caseValue), value);
+ }
+
+ /** Create a {@link Value} specifying which field to set and the value to set. */
+ public <T> Value createValue(int caseValue, T value) {
+ return createValue(getCaseEnumType().valueOf(caseValue), value);
  }
 
  /** Create a {@link Value} specifying which field to set and the value to set. */

sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoDomain.java

@@ -163,15 +163,15 @@ private void indexDescriptorByName() {
  .values()
  .forEach(
  fileDescriptor -> {
- fileDescriptor
- .getMessageTypes()
- .forEach(
- descriptor -> {
- descriptorMap.put(descriptor.getFullName(), descriptor);
- });
+ fileDescriptor.getMessageTypes().forEach(descriptor -> indexDescriptor(descriptor));
  });
  }
 
+ private void indexDescriptor(Descriptors.Descriptor descriptor) {
+ descriptorMap.put(descriptor.getFullName(), descriptor);
+ descriptor.getNestedTypes().forEach(nested -> indexDescriptor(nested));
+ }
+
  private void indexOptionsByNumber(Collection<Descriptors.FileDescriptor> fileDescriptors) {
  fieldOptionMap = new HashMap<>();
  fileOptionMap = new HashMap<>();