static void extractSchemaWithComplexTypeHandling(
      Descriptors.FieldDescriptor fieldSchema,
      List<String> fieldsToUnnest,
      String delimiter,
      String path,
      Schema pinotSchema,
      @Nullable Map<String, FieldSpec.FieldType> fieldTypeMap,
      @Nullable TimeUnit timeUnit) {
    Descriptors.FieldDescriptor.Type fieldType = fieldSchema.getType();
    if (fieldSchema.isRepeated()) {
      if (isPrimitiveType(fieldType)) {
        addFieldToPinotSchema(pinotSchema, valueOf(fieldType), path, false, fieldTypeMap, timeUnit);
      } else if (fieldsToUnnest.contains(path) && !fieldSchema.isMapField()) {
        for (Descriptors.FieldDescriptor innerField : fieldSchema.getMessageType().getFields()) {
          extractSchemaWithComplexTypeHandling(innerField, fieldsToUnnest, delimiter,
              String.join(delimiter, path, innerField.getName()), pinotSchema, fieldTypeMap, timeUnit);
        }
      } else if (!fieldSchema.isMapField()) {
        addFieldToPinotSchema(pinotSchema, FieldSpec.DataType.STRING, path, true, fieldTypeMap, timeUnit);
      }
      // Ignores Map type since it's not supported when complex type handling is enabled
    } else if (fieldType == Descriptors.FieldDescriptor.Type.MESSAGE) {
      for (Descriptors.FieldDescriptor innerField : fieldSchema.getMessageType().getFields()) {
        extractSchemaWithComplexTypeHandling(innerField, fieldsToUnnest, delimiter,
            String.join(delimiter, path, innerField.getName()), pinotSchema, fieldTypeMap, timeUnit);
      }
    } else {
      FieldSpec.DataType dataType = valueOf(fieldType);
      addFieldToPinotSchema(pinotSchema, dataType, path, true, fieldTypeMap, timeUnit);
    }
  }
@Test
  public void testExtractSchemaWithComplexTypeHandlingNestedMessage() {
    Descriptors.FieldDescriptor desc = ComplexTypes.TestMessage.getDescriptor().findFieldByName(NESTED_MESSAGE);
    Schema schema = new Schema();
    ProtoBufSchemaUtils.extractSchemaWithComplexTypeHandling(
        desc,
        Collections.emptyList(),
        ".",
        desc.getName(),
        schema,
        new HashMap<>(),
        TimeUnit.SECONDS);
    Schema expectedSchema = new Schema.SchemaBuilder()
          .addSingleValueDimension("nested_message.nested_string_field", FieldSpec.DataType.STRING)
          .addSingleValueDimension("nested_message.nested_int_field", FieldSpec.DataType.INT)
          .build();
    assertEquals(expectedSchema, schema);

    schema = new Schema();
    ProtoBufSchemaUtils.extractSchemaWithComplexTypeHandling(
        desc,
        Collections.emptyList(),
        "__",
        desc.getName(),
        schema,
        new HashMap<>(),
        TimeUnit.SECONDS);
    expectedSchema = new Schema.SchemaBuilder()
        .addSingleValueDimension("nested_message__nested_string_field", FieldSpec.DataType.STRING)
        .addSingleValueDimension("nested_message__nested_int_field", FieldSpec.DataType.INT)
        .build();
    assertEquals(expectedSchema, schema);


    desc = ComplexTypes.TestMessage.getDescriptor().findFieldByName(REPEATED_NESTED_MESSAGES);
    schema = new Schema();
    ProtoBufSchemaUtils.extractSchemaWithComplexTypeHandling(
        desc,
        Collections.emptyList(),
        "__",
        desc.getName(),
        schema,
        new HashMap<>(),
        TimeUnit.SECONDS);
    expectedSchema = new Schema.SchemaBuilder()
        .addSingleValueDimension(REPEATED_NESTED_MESSAGES, FieldSpec.DataType.STRING)
        .build();
    assertEquals(expectedSchema, schema);

    schema = new Schema();
    ProtoBufSchemaUtils.extractSchemaWithComplexTypeHandling(
        desc,
        Collections.singletonList(REPEATED_NESTED_MESSAGES),
        "__",
        desc.getName(),
        schema,
        new HashMap<>(),
        TimeUnit.SECONDS);
    expectedSchema = new Schema.SchemaBuilder()
        .addSingleValueDimension("repeated_nested_messages__nested_string_field", FieldSpec.DataType.STRING)
        .addSingleValueDimension("repeated_nested_messages__nested_int_field", FieldSpec.DataType.INT)
        .build();
    assertEquals(expectedSchema, schema);
  }