[ZEPPELIN-2257] notification about incompleteness of data

jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCInterpreter.java

@@ -47,6 +47,7 @@
 import org.apache.zeppelin.interpreter.InterpreterException;
 import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.apache.zeppelin.interpreter.InterpreterResult.Code;
+import org.apache.zeppelin.interpreter.ResultMessages;
 import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
 import org.apache.zeppelin.jdbc.security.JDBCSecurityImpl;
 import org.apache.zeppelin.scheduler.Scheduler;
@@ -474,7 +475,7 @@ private String getResults(ResultSet resultSet, boolean isTableType)
     msg.append(NEWLINE);
 
     int displayRowCount = 0;
-    while (resultSet.next() && displayRowCount < getMaxResult()) {
+    while (displayRowCount < getMaxResult() && resultSet.next()) {
       for (int i = 1; i < md.getColumnCount() + 1; i++) {
         Object resultObject;
         String resultValue;
@@ -602,8 +603,13 @@ private InterpreterResult executeSql(String propertyKey, String sql,
               interpreterResult.add(InterpreterResult.Type.TEXT,
                   "Query executed successfully.");
             } else {
-              interpreterResult.add(
-                  getResults(resultSet, !containsIgnoreCase(sqlToExecute, EXPLAIN_PREDICATE)));
+              String results = getResults(resultSet,
+                  !containsIgnoreCase(sqlToExecute, EXPLAIN_PREDICATE));
+              interpreterResult.add(results);
+              if (resultSet.next()) {
+                interpreterResult.add(ResultMessages.getExceedsLimitRowsMessage(getMaxResult(),
+                    String.format("%s.%s", COMMON_KEY, MAX_LINE_KEY)));
+              }
             }
           } else {
             // Response contains either an update count or there are no results.

jdbc/src/test/java/org/apache/zeppelin/jdbc/JDBCInterpreterTest.java

@@ -254,6 +254,8 @@ public void testSelectQueryMaxResult() throws SQLException, IOException {
     assertEquals(InterpreterResult.Code.SUCCESS, interpreterResult.code());
     assertEquals(InterpreterResult.Type.TABLE, interpreterResult.message().get(0).getType());
     assertEquals("ID\tNAME\na\ta_name\n", interpreterResult.message().get(0).getData());
+    assertEquals(InterpreterResult.Type.HTML, interpreterResult.message().get(1).getType());
+    assertTrue(interpreterResult.message().get(1).getData().contains("alert-warning"));
   }
 
   @Test

livy/src/main/java/org/apache/zeppelin/livy/LivySparkSQLInterpreter.java

@@ -141,8 +141,8 @@ public InterpreterResult interpret(String line, InterpreterContext context) {
             List<String> rows = parseSQLOutput(message.getData());
             result2.add(InterpreterResult.Type.TABLE, StringUtils.join(rows, "\n"));
             if (rows.size() >= (maxResult + 1)) {
-              result2.add(InterpreterResult.Type.HTML,
-                  "<font color=red>Results are limited by " + maxResult + ".</font>");
+              result2.add(ResultMessages.getExceedsLimitRowsMessage(maxResult,
+                  ZEPPELIN_LIVY_SPARK_SQL_MAX_RESULT));
             }
           } else {
             result2.add(message.getType(), message.getData());

pig/src/main/java/org/apache/zeppelin/pig/PigQueryInterpreter.java

@@ -45,6 +45,7 @@
 public class PigQueryInterpreter extends BasePigInterpreter {
 
   private static Logger LOGGER = LoggerFactory.getLogger(PigQueryInterpreter.class);
+  private static final String MAX_RESULTS = "zeppelin.pig.maxResult";
   private PigServer pigServer;
   private int maxResult;
 
@@ -55,7 +56,7 @@ public PigQueryInterpreter(Properties properties) {
   @Override
   public void open() {
     pigServer = getPigInterpreter().getPigServer();
-    maxResult = Integer.parseInt(getProperty("zeppelin.pig.maxResult"));
+    maxResult = Integer.parseInt(getProperty(MAX_RESULTS));
   }
 
   @Override
@@ -104,7 +105,7 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
       Iterator<Tuple> iter = pigServer.openIterator(alias);
       boolean firstRow = true;
       int index = 0;
-      while (iter.hasNext() && index <= maxResult) {
+      while (iter.hasNext() && index < maxResult) {
         index++;
         Tuple tuple = iter.next();
         if (firstRow && !schemaKnown) {
@@ -118,7 +119,8 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
         resultBuilder.append("\n");
       }
       if (index >= maxResult && iter.hasNext()) {
-        resultBuilder.append("\n<font color=red>Results are limited by " + maxResult + ".</font>");
+        resultBuilder.append("\n");
+        resultBuilder.append(ResultMessages.getExceedsLimitRowsMessage(maxResult, MAX_RESULTS));
       }
     } catch (IOException e) {
       // Extract error in the following order

pig/src/test/java/org/apache/zeppelin/pig/PigQueryInterpreterTest.java

@@ -153,6 +153,6 @@ public void testMaxResult() throws IOException {
     assertEquals(InterpreterResult.Type.TABLE, result.message().get(0).getType());
     assertEquals(InterpreterResult.Code.SUCCESS, result.code());
     assertTrue(result.message().get(0).getData().contains("id\n0\n1\n2"));
-    assertTrue(result.message().get(0).getData().contains("Results are limited by 20"));
+    assertTrue(result.message().get(1).getData().contains("alert-warning"));
   }
 }

spark/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java

@@ -42,7 +42,10 @@
  * Spark SQL interpreter for Zeppelin.
  */
 public class SparkSqlInterpreter extends Interpreter {
-  Logger logger = LoggerFactory.getLogger(SparkSqlInterpreter.class);
+  private Logger logger = LoggerFactory.getLogger(SparkSqlInterpreter.class);
+
+  public static final String MAX_RESULTS = "zeppelin.spark.maxResult";
+
   AtomicInteger num = new AtomicInteger(0);
 
   private int maxResult;
@@ -53,7 +56,7 @@ public SparkSqlInterpreter(Properties property) {
 
   @Override
   public void open() {
-    this.maxResult = Integer.parseInt(getProperty("zeppelin.spark.maxResult"));
+    this.maxResult = Integer.parseInt(getProperty(MAX_RESULTS));
   }
 
   private SparkInterpreter getSparkInterpreter() {

spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java

@@ -46,6 +46,7 @@
 import org.apache.zeppelin.interpreter.InterpreterException;
 import org.apache.zeppelin.interpreter.InterpreterHookRegistry;
 import org.apache.zeppelin.interpreter.RemoteWorksController;
+import org.apache.zeppelin.interpreter.ResultMessages;
 import org.apache.zeppelin.interpreter.remote.RemoteEventClientWrapper;
 import org.apache.zeppelin.spark.dep.SparkDependencyResolver;
 import org.apache.zeppelin.resource.Resource;
@@ -295,9 +296,9 @@ public static String showDF(SparkContext sc,
     }
 
     if (rows.length > maxResult) {
-      msg.append("<!--TABLE_COMMENT-->");
       msg.append("\n");
-      msg.append("<font color=red>Results are limited by " + maxResult + ".</font>");
+      msg.append(ResultMessages.getExceedsLimitRowsMessage(maxResult,
+          SparkSqlInterpreter.MAX_RESULTS));
     }
     sc.clearJobGroup();
     return msg.toString();

spark/src/test/java/org/apache/zeppelin/spark/SparkSqlInterpreterTest.java

@@ -47,7 +47,7 @@ public class SparkSqlInterpreterTest {
   public static void setUp() throws Exception {
     Properties p = new Properties();
     p.putAll(SparkInterpreterTest.getSparkTestProperties(tmpDir));
-    p.setProperty("zeppelin.spark.maxResult", "1000");
+    p.setProperty("zeppelin.spark.maxResult", "10");
     p.setProperty("zeppelin.spark.concurrentSQL", "false");
     p.setProperty("zeppelin.spark.sql.stacktrace", "false");
 
@@ -160,4 +160,21 @@ public void test_null_value_in_row() {
     assertEquals(Type.TABLE, ret.message().get(0).getType());
     assertEquals("name\tage\ngates\tnull\n", ret.message().get(0).getData());
   }
+
+  @Test
+  public void testMaxResults() {
+    repl.interpret("case class P(age:Int)", context);
+    repl.interpret(
+        "val gr = sc.parallelize(Seq(P(1),P(2),P(3),P(4),P(5),P(6),P(7),P(8),P(9),P(10),P(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());
+    assertTrue(ret.message().get(1).getData().contains("alert-warning"));
+  }
 }

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterOutput.java

@@ -184,9 +184,9 @@ public void write(int b) throws IOException {
         if (b == NEW_LINE_CHAR && currentOut != null) {
           InterpreterResult.Type type = currentOut.getType();
           if (type == InterpreterResult.Type.TEXT || type == InterpreterResult.Type.TABLE) {
-
-            setType(InterpreterResult.Type.TEXT);
-            getCurrentOutput().write("Output exceeds " + limit + ". Truncated.\n");
+            setType(InterpreterResult.Type.HTML);
+            getCurrentOutput().write(ResultMessages.getExceedsLimitSizeMessage(limit,
+                "ZEPPELIN_INTERPRETER_OUTPUT_LIMIT").getData().getBytes());
             truncated = true;
             return;
           }

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterResult.java

@@ -96,6 +96,10 @@ public void add(Type type, String data) {
     msg.add(new InterpreterResultMessage(type, data));
   }
 
+  public void add(InterpreterResultMessage interpreterResultMessage) {
+    msg.add(interpreterResultMessage);
+  }
+
   public Code code() {
     return code;
   }

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/ResultMessages.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+/**
+ *
+ */
+public class ResultMessages {
+  public static final String EXCEEDS_LIMIT_ROWS =
+      "<strong>Output is truncated</strong> to %s rows. Learn more about <strong>%s</strong>";
+  public static final String EXCEEDS_LIMIT_SIZE =
+      "<strong>Output is truncated</strong> to %s bytes. Learn more about <strong>%s</strong>";
+  public static final String EXCEEDS_LIMIT =
+      "<div class=\"result-alert alert-warning\" role=\"alert\">" +
+          "<button type=\"button\" class=\"close\" data-dismiss=\"alert\" aria-label=\"Close\">" +
+          "<span aria-hidden=\"true\">&times;</span></button>" +
+          "%s" +
+          "</div>";
+
+  public static InterpreterResultMessage getExceedsLimitRowsMessage(int amount, String variable) {
+    InterpreterResultMessage message = new InterpreterResultMessage(InterpreterResult.Type.HTML,
+        String.format(EXCEEDS_LIMIT, String.format(EXCEEDS_LIMIT_ROWS, amount, variable)));
+    return message;
+  }
+
+  public static InterpreterResultMessage getExceedsLimitSizeMessage(int amount, String variable) {
+    InterpreterResultMessage message = new InterpreterResultMessage(InterpreterResult.Type.HTML,
+        String.format(EXCEEDS_LIMIT, String.format(EXCEEDS_LIMIT_SIZE, amount, variable)));
+    return message;
+  }
+}

zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/InterpreterOutputTest.java

@@ -171,13 +171,15 @@ public void testTruncate() throws IOException {
     // truncate text
     out.write("%text hello\nworld\n");
     assertEquals("hello", new String(out.getOutputAt(0).toByteArray()));
-    assertTrue(new String(out.getOutputAt(1).toByteArray()).contains("Truncated"));
+    out.getOutputAt(1).flush();
+    assertTrue(new String(out.getOutputAt(1).toByteArray()).contains("truncated"));
 
     // truncate table
     out = new InterpreterOutput(this);
     out.write("%table key\tvalue\nhello\t100\nworld\t200\n");
     assertEquals("key\tvalue", new String(out.getOutputAt(0).toByteArray()));
-    assertTrue(new String(out.getOutputAt(1).toByteArray()).contains("Truncated"));
+    out.getOutputAt(1).flush();
+    assertTrue(new String(out.getOutputAt(1).toByteArray()).contains("truncated"));
 
     // does not truncate html
     out = new InterpreterOutput(this);

zeppelin-web/src/app/notebook/paragraph/result/result.css

@@ -13,9 +13,9 @@
  */
 
 .result-chart-selector {
-    margin-bottom: 10px; 
-    position: relative; 
-    display: inline-block; 
+    margin-bottom: 10px;
+    position: relative;
+    display: inline-block;
     vertical-align: middle;
 }
 
@@ -40,3 +40,9 @@
   -moz-transform: rotate(90deg) scaleX(-1);
   -ms-transform: rotate(90deg) scaleX(-1);
 }
+
+.result-alert {
+  padding: 15px;
+  border: 1px solid transparent;
+  border-radius: 4px;
+}