public static Schema getPinotSchemaFromAvroSchemaWithComplexTypeHandling(org.apache.avro.Schema avroSchema,
      @Nullable Map<String, FieldSpec.FieldType> fieldTypeMap, @Nullable TimeUnit timeUnit, List<String> fieldsToUnnest,
      String delimiter, ComplexTypeConfig.CollectionNotUnnestedToJson collectionNotUnnestedToJson) {
    Schema pinotSchema = new Schema();

    for (Field field : avroSchema.getFields()) {
      extractSchemaWithComplexTypeHandling(field.schema(), fieldsToUnnest, delimiter, field.name(), pinotSchema,
          fieldTypeMap, timeUnit, collectionNotUnnestedToJson);
    }
    return pinotSchema;
  }
@Test
  public void testGetPinotSchemaFromAvroSchemaWithComplexType()
      throws IOException {
    // do not unnest collect
    org.apache.avro.Schema avroSchema =
        new org.apache.avro.Schema.Parser().parse(ClassLoader.getSystemResourceAsStream(AVRO_NESTED_SCHEMA));
    Map<String, FieldSpec.FieldType> fieldSpecMap =
        new ImmutableMap.Builder<String, FieldSpec.FieldType>().put("d1", FieldType.DIMENSION)
            .put("hoursSinceEpoch", FieldType.DATE_TIME).put("m1", FieldType.METRIC).build();
    Schema inferredPinotSchema =
        AvroUtils.getPinotSchemaFromAvroSchemaWithComplexTypeHandling(avroSchema, fieldSpecMap, TimeUnit.HOURS,
            new ArrayList<>(), ".", ComplexTypeConfig.CollectionNotUnnestedToJson.NON_PRIMITIVE);
    Schema expectedSchema =
        new Schema.SchemaBuilder().addSingleValueDimension("d1", DataType.STRING).addMetric("m1", DataType.INT)
            .addSingleValueDimension("tuple.streetaddress", DataType.STRING)
            .addSingleValueDimension("tuple.city", DataType.STRING).addSingleValueDimension("entries", DataType.STRING)
            .addMultiValueDimension("d2", DataType.INT)
            .addDateTime("hoursSinceEpoch", DataType.LONG, "1:HOURS:EPOCH", "1:HOURS").build();
    assertEquals(inferredPinotSchema, expectedSchema);

    // unnest collection entries
    inferredPinotSchema =
        AvroUtils.getPinotSchemaFromAvroSchemaWithComplexTypeHandling(avroSchema, fieldSpecMap, TimeUnit.HOURS,
            Lists.newArrayList("entries"), ".", ComplexTypeConfig.CollectionNotUnnestedToJson.NON_PRIMITIVE);
    expectedSchema =
        new Schema.SchemaBuilder().addSingleValueDimension("d1", DataType.STRING).addMetric("m1", DataType.INT)
            .addSingleValueDimension("tuple.streetaddress", DataType.STRING)
            .addSingleValueDimension("tuple.city", DataType.STRING).addSingleValueDimension("entries.id", DataType.LONG)
            .addSingleValueDimension("entries.description", DataType.STRING).addMultiValueDimension("d2", DataType.INT)
            .addDateTime("hoursSinceEpoch", DataType.LONG, "1:HOURS:EPOCH", "1:HOURS").build();
    assertEquals(inferredPinotSchema, expectedSchema);

    // change delimiter
    inferredPinotSchema =
        AvroUtils.getPinotSchemaFromAvroSchemaWithComplexTypeHandling(avroSchema, fieldSpecMap, TimeUnit.HOURS,
            Lists.newArrayList(), "_", ComplexTypeConfig.CollectionNotUnnestedToJson.NON_PRIMITIVE);
    expectedSchema =
        new Schema.SchemaBuilder().addSingleValueDimension("d1", DataType.STRING).addMetric("m1", DataType.INT)
            .addSingleValueDimension("tuple_streetaddress", DataType.STRING)
            .addSingleValueDimension("tuple_city", DataType.STRING).addSingleValueDimension("entries", DataType.STRING)
            .addMultiValueDimension("d2", DataType.INT)
            .addDateTime("hoursSinceEpoch", DataType.LONG, "1:HOURS:EPOCH", "1:HOURS").build();
    assertEquals(inferredPinotSchema, expectedSchema);

    // change the handling of collection-to-json option, d2 will become string
    inferredPinotSchema =
        AvroUtils.getPinotSchemaFromAvroSchemaWithComplexTypeHandling(avroSchema, fieldSpecMap, TimeUnit.HOURS,
            Lists.newArrayList("entries"), ".", ComplexTypeConfig.CollectionNotUnnestedToJson.ALL);
    expectedSchema =
        new Schema.SchemaBuilder().addSingleValueDimension("d1", DataType.STRING).addMetric("m1", DataType.INT)
            .addSingleValueDimension("tuple.streetaddress", DataType.STRING)
            .addSingleValueDimension("tuple.city", DataType.STRING).addSingleValueDimension("entries.id", DataType.LONG)
            .addSingleValueDimension("entries.description", DataType.STRING)
            .addSingleValueDimension("d2", DataType.STRING)
            .addDateTime("hoursSinceEpoch", DataType.LONG, "1:HOURS:EPOCH", "1:HOURS").build();
    assertEquals(inferredPinotSchema, expectedSchema);
  }