public static Schema getPinotSchemaFromPinotSchemaWithComplexTypeHandling(Descriptors.Descriptor protoSchema,
      @Nullable Map<String, FieldSpec.FieldType> fieldTypeMap, @Nullable TimeUnit timeUnit, List<String> fieldsToUnnest,
      String delimiter) {
    Schema pinotSchema = new Schema();

    for (Descriptors.FieldDescriptor field : protoSchema.getFields()) {
      extractSchemaWithComplexTypeHandling(field, fieldsToUnnest, delimiter, field.getName(), pinotSchema,
          fieldTypeMap, timeUnit);
    }
    return pinotSchema;
  }
@Test
  public void testExtractSchemaWithCompositeTypeHandlingWithTimeColumnAndFieldSpec() {
    Descriptors.Descriptor desc = CompositeTypes.CompositeMessage.getDescriptor();
    Map<String, FieldSpec.FieldType> fieldTypeMap = new HashMap<>();
    fieldTypeMap.put("test_message.long_field", FieldSpec.FieldType.DATE_TIME);
    fieldTypeMap.put("test_message.int_field", FieldSpec.FieldType.METRIC);
    fieldTypeMap.put("test_message.double_field", FieldSpec.FieldType.METRIC);
    Schema schema = ProtoBufSchemaUtils.getPinotSchemaFromPinotSchemaWithComplexTypeHandling(
        desc,
        fieldTypeMap,
        TimeUnit.SECONDS,
        Collections.emptyList(),
        ".");
    FieldSpec fieldSpec = schema.getFieldSpecFor("test_message.long_field");
    FieldSpec expectedSchema = new DateTimeFieldSpec("test_message.long_field", FieldSpec.DataType.LONG,
        "1:SECONDS:EPOCH", "1:SECONDS");
    assertEquals(expectedSchema, fieldSpec);

    fieldSpec = schema.getFieldSpecFor("test_message.int_field");
    expectedSchema = new MetricFieldSpec("test_message.int_field", FieldSpec.DataType.INT);
    assertEquals(expectedSchema, fieldSpec);

    fieldSpec = schema.getFieldSpecFor("test_message.double_field");
    expectedSchema = new MetricFieldSpec("test_message.double_field", FieldSpec.DataType.DOUBLE);
    assertEquals(expectedSchema, fieldSpec);
  }