import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Override
public InterpreterResult interpret(String gremlin, InterpreterContext interpreterContext) {
    logger.info("execute gremlin traversal {}", gremlin);
    try {
        ResultSet results = client.submit(gremlin);
        JsonArray array = results.stream()
                                 .map(result -> new JsonPrimitive(results.toString()))
                                 .collect(JsonArray::new, JsonArray::add, JsonArray::addAll);

        //TODO extract ResultSet
        //Case MessageSerializer

        return new InterpreterResult(InterpreterResult.Code.SUCCESS, array.toString());
    } catch (RuntimeException e) {
        return new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage());
    }
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testForMapPrefix() throws SQLException, IOException {
    Properties properties = new Properties();
    properties.setProperty("common.max_count", "1000");
    properties.setProperty("common.max_retry", "3");
    properties.setProperty("default.driver", "org.apache.hive.jdbc.HiveDriver");
    properties.setProperty("default.url", getJdbcConnection());
    properties.setProperty("default.user", "");
    properties.setProperty("default.password", "");
    ImpalaInterpreter t = new ImpalaInterpreter(properties);
    t.open();

    String sqlQuery = "(fake) select * from test_table";

    InterpreterResult interpreterResult = t.interpret(sqlQuery, interpreterContext);

    // if prefix not found return ERROR and Prefix not found.
    assertEquals(InterpreterResult.Code.ERROR, interpreterResult.code());
    assertEquals("Prefix not found.", interpreterResult.message());
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testSelectQuery() throws SQLException, IOException {
    Properties properties = new Properties();
    properties.setProperty("common.max_count", "1000");
    properties.setProperty("common.max_retry", "3");
    properties.setProperty("default.driver", "org.apache.hive.jdbc.HiveDriver");
    properties.setProperty("default.url", getJdbcConnection());
    properties.setProperty("default.user", "");
    properties.setProperty("default.password", "");
    ImpalaInterpreter t = new ImpalaInterpreter(properties);
    t.open();

    String sqlQuery = "select * from test_table WHERE ID in ('a', 'b')";

    InterpreterResult interpreterResult = t.interpret(sqlQuery, interpreterContext);

    assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
    assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type());
    assertEquals("id\tname\na\ta_name\nb\tb_name\n", interpreterResult.message());

}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Override
public InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter) {
  String propertyKey = getPropertyKey(cmd);

  if (null != propertyKey) {
    cmd = cmd.substring(propertyKey.length() + 2);
  } else {
    propertyKey = DEFAULT_KEY;
  }

  cmd = cmd.trim();

  logger.info("PropertyKey: {}, SQL command: '{}'", propertyKey, cmd);

  return executeSql(propertyKey, cmd, contextInterpreter);
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void jdbcRestart() throws IOException, SQLException, ClassNotFoundException {
  Properties properties = new Properties();
  properties.setProperty("common.max_count", "1000");
  properties.setProperty("common.max_retry", "3");
  properties.setProperty("default.driver", "org.h2.Driver");
  properties.setProperty("default.url", getJdbcConnection());
  properties.setProperty("default.user", "");
  properties.setProperty("default.password", "");
  HiveInterpreter t = new HiveInterpreter(properties);
  t.open();

  InterpreterResult interpreterResult =
      t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null));
  assertEquals("ID\tNAME\na\ta_name\nb\tb_name\n", interpreterResult.message());

  t.getConnection("default").close();

  interpreterResult =
      t.interpret("select * from test_table", new InterpreterContext("", "1", "","", null,null,null,null));
  assertEquals("ID\tNAME\na\ta_name\nb\tb_name\n", interpreterResult.message());
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void should_describe_udt_withing_logged_in_keyspace() throws Exception {
    //Given
    String query = "USE live_data;\n" +
            "DESCRIBE TYPE address;";
    final String expected = reformatHtml(
            readTestResource("/scalate/DescribeType_live_data_address_within_current_keyspace.html"));

    //When
    final InterpreterResult actual = interpreter.interpret(query, intrContext);

    //Then
    assertThat(actual.code()).isEqualTo(Code.SUCCESS);

    assertThat(reformatHtml(actual.message())).isEqualTo(expected);
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
/**
 * Clone paragraph and add it to note.
 *
 * @param srcParagraph
 */
public void addCloneParagraph(Paragraph srcParagraph) {
  Paragraph newParagraph = new Paragraph(this, this, replLoader);

  Map<String, Object> config = new HashMap<>(srcParagraph.getConfig());
  Map<String, Object> param = new HashMap<>(srcParagraph.settings.getParams());
  Map<String, Input> form = new HashMap<>(srcParagraph.settings.getForms());
  Gson gson = new Gson();
  InterpreterResult result = gson.fromJson(
      gson.toJson(srcParagraph.getReturn()),
      InterpreterResult.class);

  newParagraph.setConfig(config);
  newParagraph.settings.setParams(param);
  newParagraph.settings.setForms(form);
  newParagraph.setText(srcParagraph.getText());
  newParagraph.setTitle(srcParagraph.getTitle());
  newParagraph.setReturn(result, null);

  synchronized (paragraphs) {
    paragraphs.add(newParagraph);
  }
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
private InterpreterResult HandleHelp(JLineShell shell, String st) {
  java.util.logging.StreamHandler sh = null;
  java.util.logging.Logger springLogger = null;
  java.util.logging.Formatter formatter = new java.util.logging.Formatter() {
    public String format(java.util.logging.LogRecord record) {
      return record.getMessage();
    }
  };
  ByteArrayOutputStream baos = new ByteArrayOutputStream();
  try {
    sh = new java.util.logging.StreamHandler(baos, formatter);
    springLogger = HandlerUtils.getLogger(org.springframework.shell.core.SimpleParser.class);
    springLogger.addHandler(sh);
    shell.executeCommand(st);
  } catch (Exception e) {
    s_logger.error(e.getMessage(), e);
    return new InterpreterResult(Code.ERROR, e.getMessage());
  }
  finally {
    sh.flush();
    springLogger.removeHandler(sh);
    sh.close();
  }
  return new InterpreterResult(Code.SUCCESS, baos.toString());
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
private InterpreterResult executeQuery(String sql) throws IOException {

    HttpResponse response = prepareRequest(sql);

    if (response.getStatusLine().getStatusCode() != 200) {
      logger.error("failed to execute query: " + response.getEntity().getContent().toString());
      return new InterpreterResult(InterpreterResult.Code.ERROR,
          "Failed : HTTP error code " + response.getStatusLine().getStatusCode());
    }

    BufferedReader br = new BufferedReader(
        new InputStreamReader((response.getEntity().getContent())));
    StringBuilder sb = new StringBuilder();

    String output;
    logger.info("Output from Server .... \n");
    while ((output = br.readLine()) != null) {
      logger.info(output);
      sb.append(output).append('\n');
    }
    InterpreterResult rett = new InterpreterResult(InterpreterResult.Code.SUCCESS, 
        formatResult(sb.toString()));
    return rett;
  }
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void test() {
  Properties prop = new Properties();
  prop.setProperty(LENS_SERVER_URL, "http://127.0.0.1:9999/lensapi");
  prop.setProperty(LENS_CLIENT_DBNAME, "default");
  prop.setProperty(LENS_PERSIST_RESULTSET, "false");
  prop.setProperty(LENS_SESSION_CLUSTER_USER, "default");
  prop.setProperty(ZEPPELIN_MAX_ROWS, "1000");
  prop.setProperty(ZEPPELIN_LENS_RUN_CONCURRENT_SESSION, "true");
  prop.setProperty(ZEPPELIN_LENS_CONCURRENT_SESSIONS, "10");
  LensInterpreter t = new MockLensInterpreter(prop);
  t.open();
  //simple help test
  InterpreterResult result = t.interpret("help", null);
  assertEquals(result.type(), InterpreterResult.Type.TEXT);
  //assertEquals("unable to find 'query execute' in help message", 
  //  result.message().contains("query execute"), result.message());
  t.close();
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testGraphQueries() {
    InterpreterResult res = interpreter.interpret("FOR x IN actsIn COLLECT actor = x._from WITH COUNT INTO counter FILTER counter >= 2 RETURN {actor: actor, movies: counter}", null);
    assertEquals(Code.SUCCESS, res.code());

    res = interpreter.interpret("FOR x IN actsIn COLLECT movie = x._to WITH COUNT INTO counter FILTER counter == 1 RETURN movie", null);
    assertEquals(Code.SUCCESS, res.code());

    res = interpreter.interpret("FOR x IN actsIn COLLECT movie = x._to WITH COUNT INTO counter RETURN {movie: movie, actors: counter}", null);
    assertEquals(Code.SUCCESS, res.code());

    //res = interpreter.interpret("FOR x IN ANY 'movies/TheMatrix' actsIn OPTIONS {bfs: true, uniqueVertices: 'global'} RETURN x._id", null);
    res = interpreter.interpret("FOR x IN UNION_DISTINCT ("
        + "(FOR y IN ANY 'movies/TheMatrix' actsIn OPTIONS {bfs: true, uniqueVertices: 'global'} RETURN y._id), "
        + "(FOR y IN ANY 'movies/TheDevilsAdvocate' actsIn OPTIONS {bfs: true, uniqueVertices: 'global'} RETURN y._id)"
        + ") RETURN x", null);
    assertEquals(Code.SUCCESS, res.code());

    res = interpreter.interpret("FOR x IN INTERSECTION ("
        + "(FOR y IN ANY 'movies/TheMatrix' actsIn OPTIONS {bfs: true, uniqueVertices: 'global'} RETURN y._id), "
        + "(FOR y IN ANY 'movies/TheDevilsAdvocate' actsIn OPTIONS {bfs: true, uniqueVertices: 'global'} RETURN y._id)) RETURN x", null);
    System.out.println(res.message());
    assertEquals(Code.SUCCESS, res.code());
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testAngularObjectAddOnZeppelinServerSide() throws InterruptedException {
  // test if angularobject add from server side propagate to interpreter process's registry.
  // will happen when zeppelin server loads notebook and restore the object into registry

  InterpreterResult ret = intp.interpret("get", context);
  Thread.sleep(500); // waitFor eventpoller pool event
  String[] result = ret.message().split(" ");
  assertEquals("0", result[0]); // size of registry
  
  // create object
  localRegistry.addAndNotifyRemoteProcess("n1", "v1", "note");
  
  // get from remote registry 
  ret = intp.interpret("get", context);
  Thread.sleep(500); // waitFor eventpoller pool event
  result = ret.message().split(" ");
  assertEquals("1", result[0]); // size of registry
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testBasicIntp() {
  assertEquals(InterpreterResult.Code.SUCCESS,
      repl.interpret("val a = 1\nval b = 2", context).code());

  // when interpret incomplete expression
  InterpreterResult incomplete = repl.interpret("val a = \"\"\"", context);
  assertEquals(InterpreterResult.Code.INCOMPLETE, incomplete.code());
  assertTrue(incomplete.message().length() > 0); // expecting some error
                                                 // message
  /*
   * assertEquals(1, repl.getValue("a")); assertEquals(2, repl.getValue("b"));
   * repl.interpret("val ver = sc.version");
   * assertNotNull(repl.getValue("ver")); assertEquals("HELLO\n",
   * repl.interpret("println(\"HELLO\")").message());
   */
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void should_fail_when_executing_a_removed_prepared_statement() throws Exception {
    //Given
    String prepare_first = "@prepare[to_be_removed]=INSERT INTO zeppelin.users(login,deceased) VALUES(?,?)";
    interpreter.interpret(prepare_first, intrContext);
    String remove_prepared = "@remove_prepare[to_be_removed]\n" +
            "@bind[to_be_removed]='bind_bool'";

    //When
    final InterpreterResult actual = interpreter.interpret(remove_prepared, intrContext);

    //Then
    assertThat(actual.code()).isEqualTo(Code.ERROR);
    assertThat(actual.message()).isEqualTo("The statement 'to_be_removed' can not be bound to values. " +
            "Are you sure you did prepare it with @prepare[to_be_removed] ?");
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testStruct() {
  repl.interpret("case class Person(name:String, age:Int)", context);
  repl.interpret("case class People(group:String, person:Person)", context);
  repl.interpret(
      "val gr = sc.parallelize(Seq(People(\"g1\", Person(\"moon\",33)), People(\"g2\", Person(\"sun\",11))))",
      context);
  if (isDataFrameSupported()) {
    repl.interpret("gr.toDF.registerTempTable(\"gr\")", context);
  } else {
    repl.interpret("gr.registerTempTable(\"gr\")", context);
  }

  InterpreterResult ret = sql.interpret("select * from gr", context);
  assertEquals(InterpreterResult.Code.SUCCESS, ret.code());
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void should_describe_cluster() throws Exception {
    //Given

    String query = "DESCRIBE CLUSTER;";
    final String expected = reformatHtml(
            readTestResource("/scalate/DescribeCluster.html"));

    //When
    final InterpreterResult actual = interpreter.interpret(query, intrContext);

    //Then
    assertThat(actual.code()).isEqualTo(Code.SUCCESS);

    assertThat(reformatHtml(actual.message())).isEqualTo(expected);
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testSelectQuery() throws SQLException {

  when(phoenixInterpreter.getMaxResult()).thenReturn(1000);

  String sqlQuery = "select * from t";

  result.addColumn("col1", new String[] {"val11", "val12"});
  result.addColumn("col2", new String[] {"val21", "val22"});

  InterpreterResult interpreterResult = phoenixInterpreter.interpret(sqlQuery, null);

  assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
  assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type());
  assertEquals("col1\tcol2\nval11\tval21\nval12\tval22\n", interpreterResult.message());

  verifySQLStatementExecuted(sqlQuery);
  verifyAllResultSetsClosed();
  verifyAllStatementsClosed();
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testSelectQueryMaxResult() throws SQLException {

  when(phoenixInterpreter.getMaxResult()).thenReturn(1);

  String sqlQuery = "select * from t";

  result.addColumn("col1", new String[] {"val11", "val12"});
  result.addColumn("col2", new String[] {"val21", "val22"});

  InterpreterResult interpreterResult = phoenixInterpreter.interpret(sqlQuery, null);

  assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
  assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type());
  assertEquals("col1\tcol2\nval11\tval21\n", interpreterResult.message());

  verifySQLStatementExecuted(sqlQuery);
  verifyAllResultSetsClosed();
  verifyAllStatementsClosed();
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testSelectQueryWithSpecialCharacters() throws SQLException {

  when(phoenixInterpreter.getMaxResult()).thenReturn(1000);

  String sqlQuery = "select * from t";

  result.addColumn("co\tl1", new String[] {"val11", "va\tl1\n2"});
  result.addColumn("co\nl2", new String[] {"v\nal21", "val\t22"});

  InterpreterResult interpreterResult = phoenixInterpreter.interpret(sqlQuery, null);

  assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
  assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type());
  assertEquals("co l1\tco l2\nval11\tv al21\nva l1 2\tval 22\n", interpreterResult.message());

  verifySQLStatementExecuted(sqlQuery);
  verifyAllResultSetsClosed();
  verifyAllStatementsClosed();
}
 

import org.apache.zeppelin.interpreter.InterpreterResult; //导入依赖的package包/类
@Test
public void testNullColumnResult() throws SQLException {

  when(psqlInterpreter.getMaxResult()).thenReturn(1000);

  String sqlQuery = "select * from t";

  result.addColumn("col1", new String[] {"val11", null});
  result.addColumn("col2", new String[] {null, "val22"});

  InterpreterResult interpreterResult = psqlInterpreter.interpret(sqlQuery, null);

  assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
  assertEquals(InterpreterResult.Type.TABLE, interpreterResult.type());
  assertEquals("col1\tcol2\nval11\t\n\tval22\n", interpreterResult.message());

  verifySQLStatementExecuted(sqlQuery);
  verifyAllResultSetsClosed();
  verifyAllStatementsClosed();
}