public static Collection<RelRoot> populateSystemStreamsAndGetRelRoots(List<String> dslStmts, Config config,
      List<String> inputSystemStreams, List<String> outputSystemStreams) {
    // TODO: Get the converter factory based on the file type. Create abstraction around this.
    DslConverterFactory dslConverterFactory = new SamzaSqlDslConverterFactory();
    DslConverter dslConverter = dslConverterFactory.create(config);

    Collection<RelRoot> relRoots = dslConverter.convertDsl(String.join("\n", dslStmts));

    // RelRoot does not have sink node for Samza SQL dsl, so we can not traverse the relRoot tree to get
    // "outputSystemStreams"
    // FIXME: the snippet below does not work for Samza SQL dsl but is required for other dsls. Future fix could be
    // for samza sql to build TableModify for sink and stick it to the relRoot, so we could get output stream out of it.

    // for (RelRoot relRoot : relRoots) {
    //   SamzaSqlApplicationConfig.populateSystemStreams(relRoot.project(), inputSystemStreams, outputSystemStreams);
    // }

    // The below code is specific to Samza SQL dsl and should be removed once Samza SQL includes sink as part of
    // relRoot and the above code in uncommented.
    List<String> sqlStmts = SamzaSqlDslConverter.fetchSqlFromConfig(config);
    List<SamzaSqlQueryParser.QueryInfo> queryInfo = SamzaSqlDslConverter.fetchQueryInfo(sqlStmts);
    inputSystemStreams.addAll(queryInfo.stream().map(SamzaSqlQueryParser.QueryInfo::getSources).flatMap(Collection::stream)
          .collect(Collectors.toList()));
    outputSystemStreams.addAll(queryInfo.stream().map(SamzaSqlQueryParser.QueryInfo::getSink).collect(Collectors.toList()));

    return relRoots;
  }
Class representing the Samza SQL application config
samza-sql/src/main/java/org/apache/samza/sql/runner/SamzaSqlApplicationConfig.java