class IcebergTableSchema {

  public static IcebergTableSchema of(Class<?> clazz, List<String> partitionKeys) {
    List<Types.NestedField> tableSchemaFields = new ArrayList<>();
    Types.NestedField[] partitionSpecSchemaFields = new Types.NestedField[partitionKeys.size()];

    int schemaIndex = STARTING_SCHEMA_INDEX;
    int complexTypesIndex = STARTING_COMPLEX_TYPES_INDEX;

    for (Field field : clazz.getDeclaredFields()) {
      if (!field.isAnnotationPresent(MetastoreFieldDefinition.class)) {
        continue;
      }

      String typeSimpleName = field.getType().getSimpleName().toLowerCase();
      org.apache.iceberg.types.Type icebergType = JAVA_TO_ICEBERG_TYPE_MAP.get(typeSimpleName);

      if (icebergType == null && field.getAnnotatedType().getType() instanceof ParameterizedType) {
        Type[] actualTypeArguments = ((ParameterizedType) field.getAnnotatedType().getType()).getActualTypeArguments();
        switch (typeSimpleName) {
          case "list":
            org.apache.iceberg.types.Type listIcebergType = getGenericsType(actualTypeArguments[0]);
            icebergType = Types.ListType.ofOptional(complexTypesIndex++, listIcebergType);
            break;
          case "map":
            org.apache.iceberg.types.Type keyIcebergType = getGenericsType(actualTypeArguments[0]);
            org.apache.iceberg.types.Type valueIcebergType = getGenericsType(actualTypeArguments[1]);
            icebergType = Types.MapType.ofOptional(complexTypesIndex++, complexTypesIndex++, keyIcebergType, valueIcebergType);
            break;
          default:
            throw new IcebergMetastoreException(String.format(
              "Unexpected parametrized type for class [%s]: %s", clazz.getCanonicalName(), typeSimpleName));
        }
      }

      if (icebergType == null) {
        throw new IcebergMetastoreException(String.format(
          "Unexpected type for class [%s]: %s", clazz.getCanonicalName(), typeSimpleName));
      }

      Types.NestedField icebergField = Types.NestedField.optional(schemaIndex++, field.getName(), icebergType);

      tableSchemaFields.add(icebergField);

      int partitionIndex = partitionKeys.indexOf(field.getName());
      if (partitionIndex != -1) {
        partitionSpecSchemaFields[partitionIndex] = icebergField;
      }
    }

    if (Stream.of(partitionSpecSchemaFields).anyMatch(Objects::isNull)) {
      throw new IcebergMetastoreException(String.format(
        "Some of partition fields are missing in the class [%s]. Partition keys: %s. Partition values: %s.",
        clazz.getCanonicalName(), partitionKeys, Arrays.asList(partitionSpecSchemaFields)));
    }

    Schema tableSchema = new Schema(tableSchemaFields);
    PartitionSpec partitionSpec = buildPartitionSpec(partitionSpecSchemaFields);
    logger.debug("Constructed Iceberg table schema for class [{}]. Table schema : {}. Partition spec: {}.",
      clazz.getCanonicalName(), tableSchema, partitionSpec);
    return new IcebergTableSchema(tableSchema, partitionSpec);
  }

  public  IcebergTableSchema(Schema tableSchema, PartitionSpec partitionSpec);

}