public void translate(RelRoot relRoot, String outputSystemStream, TranslatorContext translatorContext, int queryId) {
    final RelNode node = relRoot.project();
    ScanTranslator scanTranslator =
        new ScanTranslator(sqlConfig.getSamzaRelConverters(), sqlConfig.getInputSystemStreamConfigBySource(), queryId);

    /* update input metrics */
    String queryLogicalId = String.format(TranslatorConstants.LOGSQLID_TEMPLATE, queryId);

    opId = 0;

    node.accept(new RelShuttleImpl() {

      @Override
      public RelNode visit(RelNode relNode) {
        // There should never be a TableModify in the calcite plan.
        Validate.isTrue(!(relNode instanceof TableModify));
        return super.visit(relNode);
      }

      @Override
      public RelNode visit(TableScan scan) {
        RelNode node = super.visit(scan);
        String logicalOpId = String.format(TranslatorConstants.LOGOPID_TEMPLATE, queryId, "scan", opId++);
        scanTranslator.translate(scan, queryLogicalId, logicalOpId, translatorContext, systemDescriptors,
            inputMsgStreams);
        return node;
      }

      @Override
      public RelNode visit(LogicalFilter filter) {
        RelNode node = visitChild(filter, 0, filter.getInput());
        String logicalOpId = String.format(TranslatorConstants.LOGOPID_TEMPLATE, queryId, "filter", opId++);
        new FilterTranslator(queryId).translate(filter, logicalOpId, translatorContext);
        return node;
      }

      @Override
      public RelNode visit(LogicalProject project) {
        RelNode node = super.visit(project);
        String logicalOpId = String.format(TranslatorConstants.LOGOPID_TEMPLATE, queryId, "project", opId++);
        new ProjectTranslator(queryId).translate(project, logicalOpId, translatorContext);
        return node;
      }

      @Override
      public RelNode visit(LogicalJoin join) {
        RelNode node = super.visit(join);
        String logicalOpId = String.format(TranslatorConstants.LOGOPID_TEMPLATE, queryId, "join", opId++);
        new JoinTranslator(logicalOpId, sqlConfig.getMetadataTopicPrefix(), queryId).translate(join, translatorContext);
        return node;
      }

      @Override
      public RelNode visit(LogicalAggregate aggregate) {
        RelNode node = super.visit(aggregate);
        String logicalOpId = String.format(TranslatorConstants.LOGOPID_TEMPLATE, queryId, "window", opId++);
        new LogicalAggregateTranslator(logicalOpId, sqlConfig.getMetadataTopicPrefix()).translate(aggregate,
            translatorContext);
        return node;
      }
    });

    String logicalOpId = String.format(TranslatorConstants.LOGOPID_TEMPLATE, queryId, "insert", opId);
    sendToOutputStream(queryLogicalId, logicalOpId, outputSystemStream, streamAppDescriptor, translatorContext, node,
        queryId);
  }
Translate Calcite plan to Samza stream operators.
@param relRoot Calcite plan in the form of {@link RelRoot}. RelRoot should not include the sink ({@link TableModify})
@param outputSystemStream Sink associated with the Calcite plan.
@param translatorContext Context maintained across translations.
@param queryId query index of the sql statement corresponding to the Calcite plan in multi SQL statement scenario
               starting with index 0.
samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java