private void sendToOutputStream(String queryLogicalId, String logicalOpId, String sinkStream,
      StreamApplicationDescriptor appDesc, TranslatorContext translatorContext, RelNode node, int queryId) {
    SqlIOConfig sinkConfig = sqlConfig.getOutputSystemStreamConfigsBySource().get(sinkStream);
    MessageStream<SamzaSqlRelMessage> stream = translatorContext.getMessageStream(node.getId());
    MessageStream<KV<Object, Object>> outputStream =
        stream.map(new OutputMapFunction(queryLogicalId, logicalOpId, sinkStream, queryId));
    Optional<TableDescriptor> tableDescriptor = sinkConfig.getTableDescriptor();
    if (!tableDescriptor.isPresent()) {
      KVSerde<Object, Object> noOpKVSerde = KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>());
      String systemName = sinkConfig.getSystemName();
      DelegatingSystemDescriptor sd = systemDescriptors.computeIfAbsent(systemName, DelegatingSystemDescriptor::new);
      GenericOutputDescriptor<KV<Object, Object>> osd = sd.getOutputDescriptor(sinkConfig.getStreamId(), noOpKVSerde);
      OutputStream stm = outputMsgStreams.computeIfAbsent(sinkConfig.getSource(), v -> appDesc.getOutputStream(osd));
      outputStream.sendTo(stm);

      // Process system events only if the output is a stream.
      if (sqlConfig.isProcessSystemEvents()) {
        for (MessageStream<SamzaSqlInputMessage> inputStream : inputMsgStreams.values()) {
          MessageStream<KV<Object, Object>> systemEventStream =
              inputStream.filter(message -> message.getMetadata().isSystemMessage())
                  .map(SamzaSqlInputMessage::getKeyAndMessageKV);

          systemEventStream.sendTo(stm);
        }
      }
    } else {
      Table outputTable = appDesc.getTable(tableDescriptor.get());
      if (outputTable == null) {
        String msg = "Failed to obtain table descriptor of " + sinkConfig.getSource();
        throw new SamzaException(msg);
      }
      outputStream.sendTo(outputTable);
    }
  }
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