private static GenericRecord convertToGenericRecord(SamzaSqlRelRecord relRecord, Schema schema) {
    GenericRecord record = new GenericData.Record(schema);
    List<String> fieldNames = relRecord.getFieldNames();
    List<Object> values = relRecord.getFieldValues();
    for (int index = 0; index < fieldNames.size(); index++) {
      if (!fieldNames.get(index).equalsIgnoreCase(SamzaSqlRelMessage.KEY_NAME)) {
        String fieldName = fieldNames.get(index);
        /**
         * It is possible that the destination Avro schema doesn't have all the fields that are projected from the
         * SQL. This is especially possible in SQL statements like
         *        insert into kafka.outputTopic select id, company from profile
         * where company is an avro record in itself whose schema can evolve. When this happens we will end up with
         * fields in the SamzaSQLRelRecord for company field which doesn't have equivalent fields in the outputTopic's schema
         * for company. To support this scenario where the input schemas and output schemas can evolve in their own cadence,
         * We ignore the fields which doesn't have corresponding schema in the output topic.
         */
        if (schema.getField(fieldName) == null) {
          LOG.debug("Schema with Name {} and Namespace {} doesn't contain the fieldName {}, Skipping it.",
              schema.getName(), schema.getNamespace(), fieldName);
          continue;
        }
        Object relObj = values.get(index);
        Schema fieldSchema = schema.getField(fieldName).schema();
        record.put(fieldName, convertToAvroObject(relObj, getNonNullUnionSchema(fieldSchema)));
      }
    }

    return record;
  }
Convert the nested relational message to the output samza message.
samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java