public static QueryInfo parseQuery(String sql) {
    Planner planner = createPlanner();
    SqlNode sqlNode;
    // Having semi-colons at the end of sql statement is a valid syntax in standard sql but not for Calcite parser.
    // Hence, removing trailing semi-colon before passing sql statement to Calcite parser.
    sql = sql.replaceAll(TRAILING_SEMI_COLON_REGEX, "");
    try {
      sqlNode = planner.parse(sql);
    } catch (SqlParseException e) {
      String errorMsg = SamzaSqlValidator.formatErrorString(sql, e);
      LOG.error(errorMsg, e);
      throw new SamzaException(errorMsg, e);
    }

    String sink;
    String selectQuery;
    ArrayList<String> sources;
    if (sqlNode instanceof SqlInsert) {
      SqlInsert sqlInsert = (SqlInsert) sqlNode;
      sink = sqlInsert.getTargetTable().toString();
      if (sqlInsert.getSource() instanceof SqlSelect) {
        SqlSelect sqlSelect = (SqlSelect) sqlInsert.getSource();
        selectQuery = sqlSelect.toString();
        LOG.info("Parsed select query {} from sql {}", selectQuery, sql);
        sources = getSourcesFromSelectQuery(sqlSelect);
      } else {
        String msg = String.format("Sql query is not of the expected format. Select node expected, found %s",
            sqlInsert.getSource().getClass().toString());
        LOG.error(msg);
        throw new SamzaException(msg);
      }
    } else {
      String msg = String.format("Sql query is not of the expected format. Insert node expected, found %s",
          sqlNode.getClass().toString());
      LOG.error(msg);
      throw new SamzaException(msg);
    }

    return new QueryInfo(selectQuery, sources, sink, sql);
  }
Utility class that is used to parse the Samza sql query to figure out the sources, sink etc..
samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java