@Override
    public SamzaSqlRelMessage apply(SamzaSqlInputMessage samzaSqlInputMessage) {
      long startProcessingMs = System.currentTimeMillis();
      long startProcessingNs = System.nanoTime();
      /* SAMZA-2089/LISAMZA-10654: the SamzaRelConverter.convertToRelMessage currently does not initialize
       *                           the samzaSqlRelMessage.samzaSqlRelMsgMetadata, this needs to be fixed */
      SamzaSqlRelMessage retMsg = this.msgConverter.convertToRelMessage(samzaSqlInputMessage.getKeyAndMessageKV());
      retMsg.setEventTime(samzaSqlInputMessage.getMetadata().getEventTime());
      retMsg.setArrivalTime(samzaSqlInputMessage.getMetadata().getArrivalTime());
      retMsg.setScanTime(startProcessingNs, startProcessingMs);
      updateMetrics(startProcessingNs, System.nanoTime());
      return retMsg;
    }
ScanMapFUnction implements MapFunction to process input SamzaSqlRelMessages into output
SamzaSqlRelMessage, performing the table scan
samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java