private MessageStream<SamzaSqlRelMessage> translateFlatten(Integer flattenIndex,
      MessageStream<SamzaSqlRelMessage> inputStream) {
    return inputStream.flatMap(message -> {
      Object targetFlattenColumn = message.getSamzaSqlRelRecord().getFieldValues().get(flattenIndex);
      final List<SamzaSqlRelMessage> outMessages = new ArrayList<>();
      if (targetFlattenColumn != null && targetFlattenColumn instanceof List) {
        List<Object> objectList = (List<Object>) targetFlattenColumn;
        SamzaSqlRelMsgMetadata messageMetadata = message.getSamzaSqlRelMsgMetadata();
        SamzaSqlRelMsgMetadata newMetadata =
            new SamzaSqlRelMsgMetadata(messageMetadata.getEventTime(), messageMetadata.getArrivalTime(),
                messageMetadata.getScanTimeNanos(), messageMetadata.getScanTimeMillis());
        for (Object fieldValue : objectList) {
          List<Object> newValues = new ArrayList<>(message.getSamzaSqlRelRecord().getFieldValues());
          newValues.set(flattenIndex, fieldValue);
          outMessages.add(
              new SamzaSqlRelMessage(message.getSamzaSqlRelRecord().getFieldNames(), newValues, newMetadata));
          newMetadata = new SamzaSqlRelMsgMetadata(newMetadata.getEventTime(), newMetadata.getArrivalTime(),
              newMetadata.getScanTimeNanos(), newMetadata.getScanTimeMillis());
        }
      } else {
        message.getSamzaSqlRelMsgMetadata().isNewInputMessage = true;
        outMessages.add(message);
      }
      return outMessages;
    });
  }
Updates the Diagnostics Metrics (processing time and number of events)
@param arrivalTime input message arrival time (= beging of processing in this operator)
@param outputTime output message output time (=end of processing in this operator)
@param isNewInputMessage whether the input Message is from new input message or not
samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java