@Override
    public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) {
      long arrivalTime = System.nanoTime();
      RelDataType type = project.getRowType();
      Object[] output = new Object[type.getFieldCount()];
      try {
        expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(),
            message.getSamzaSqlRelRecord().getFieldValues().toArray(), output);
      } catch (Exception e) {
        String errMsg = String.format("Handling the following rel message ran into an error. %s", message);
        LOG.error(errMsg, e);
        throw new SamzaException(errMsg, e);
      }
      List<String> names = new ArrayList<>();
      for (int index = 0; index < output.length; index++) {
        names.add(index, project.getNamedProjects().get(index).getValue());
      }
      updateMetrics(arrivalTime, System.nanoTime(), message.getSamzaSqlRelMsgMetadata().isNewInputMessage);
      return new SamzaSqlRelMessage(names, Arrays.asList(output), message.getSamzaSqlRelMsgMetadata());
    }
transforms the input message into the output message with projected fields
@param message  the input message to be transformed
@return the new SamzaSqlRelMessage message
samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java