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(dataProvider = "scalarCases")
  public void testExtractSchemaWithCompositeTypeHandling(
      String fieldName, FieldSpec.DataType type, boolean isSingleValue) {
    Descriptors.Descriptor desc = CompositeTypes.CompositeMessage.getDescriptor();
    FieldSpec schema = ProtoBufSchemaUtils.getPinotSchemaFromPinotSchemaWithComplexTypeHandling(
        desc,
        new HashMap<>(),
        TimeUnit.SECONDS,
        Collections.emptyList(),
        ".").getFieldSpecFor("test_message." + fieldName);
    FieldSpec expectedSchema = new DimensionFieldSpec("test_message." + fieldName, type, isSingleValue);
    assertEquals(expectedSchema, schema);
  }