void translate(final TableScan tableScan, final String queryLogicalId, final String logicalOpId,
      final TranslatorContext context, Map<String, DelegatingSystemDescriptor> systemDescriptors,
      Map<String, MessageStream<SamzaSqlInputMessage>> inputMsgStreams) {
    StreamApplicationDescriptor streamAppDesc = context.getStreamAppDescriptor();
    List<String> tableNameParts = tableScan.getTable().getQualifiedName();
    String sourceName = SqlIOConfig.getSourceFromSourceParts(tableNameParts);

    Validate.isTrue(relMsgConverters.containsKey(sourceName), String.format("Unknown source %s", sourceName));
    SqlIOConfig sqlIOConfig = systemStreamConfig.get(sourceName);
    final String systemName = sqlIOConfig.getSystemName();
    final String streamId = sqlIOConfig.getStreamId();
    final String source = sqlIOConfig.getSource();

    final boolean isRemoteTable = sqlIOConfig.getTableDescriptor().isPresent() && (
        sqlIOConfig.getTableDescriptor().get() instanceof RemoteTableDescriptor || sqlIOConfig.getTableDescriptor()
            .get() instanceof CachingTableDescriptor);

    // For remote table, we don't have an input stream descriptor. The table descriptor is already defined by the
    // SqlIOResolverFactory.
    // For local table, even though table descriptor is already defined, we still need to create the input stream
    // descriptor to load the local table.
    if (isRemoteTable) {
      return;
    }

    // set the wrapper input transformer (SamzaSqlInputTransformer) in system descriptor
    DelegatingSystemDescriptor systemDescriptor = systemDescriptors.get(systemName);
    if (systemDescriptor == null) {
      systemDescriptor = new DelegatingSystemDescriptor(systemName, new SamzaSqlInputTransformer());
      systemDescriptors.put(systemName, systemDescriptor);
    } else {
      /* in SamzaSQL, there should be no systemDescriptor setup by user, so this branch happens only
       * in case of Fan-OUT (i.e., same input stream used in multiple sql statements), or when same input
       * used twice in same sql statement (e.g., select ... from input as i1, input as i2 ...), o.w., throw error */
      if (systemDescriptor.getTransformer().isPresent()) {
        InputTransformer existingTransformer = systemDescriptor.getTransformer().get();
        if (!(existingTransformer instanceof SamzaSqlInputTransformer)) {
          throw new SamzaException(
              "SamzaSQL Exception: existing transformer for " + systemName + " is not SamzaSqlInputTransformer");
        }
      }
    }

    InputDescriptor inputDescriptor = systemDescriptor.getInputDescriptor(streamId, new NoOpSerde<>());

    if (!inputMsgStreams.containsKey(source)) {
      MessageStream<SamzaSqlInputMessage> inputMsgStream = streamAppDesc.getInputStream(inputDescriptor);
      inputMsgStreams.put(source, inputMsgStream.map(new SystemMessageMapperFunction(source, queryId)));
    }
    MessageStream<SamzaSqlRelMessage> samzaSqlRelMessageStream = inputMsgStreams.get(source)
        .filter(new FilterSystemMessageFunction(sourceName, queryId))
        .map(new ScanMapFunction(sourceName, queryId, queryLogicalId, logicalOpId));

    context.registerMessageStream(tableScan.getId(), samzaSqlRelMessageStream);
  }
Updates the MetricsRegistery of this operator
@param startProcessingNs = begin processing of the message
@param endProcessing = end of processing
samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java