code: private void initRecordBuilderFactories() {
      for (FieldMapping fieldMapping : avroSchema.getColumnMappingDescriptor().getFieldMappings()) {
        if (fieldMapping.getMappingType() == MappingType.KEY_AS_COLUMN) {
          String fieldName = fieldMapping.getFieldName();
          Schema fieldSchema = avroSchema.getAvroSchema().getField(fieldName)
              .schema();
          Schema.Type fieldSchemaType = fieldSchema.getType();
          if (fieldSchemaType == Schema.Type.RECORD) {
            AvroRecordBuilderFactory<E> factory = buildAvroRecordBuilderFactory(fieldSchema);
            kacRecordBuilderFactories.put(fieldName, factory);
          }
        }
      }
    }
docs: Initialize the AvroRecordBuilderFactories for all keyAsColumn mapped fields
  that are record types. We need to be able to get record builders for these
  since the records are broken across many columns, and need to be
  constructed by the composer.