ZEPPELIN-140 Send complete buffer to interpreter.completion

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

@@ -40,6 +40,7 @@
 import org.apache.zeppelin.interpreter.WrappedInterpreter;
 import org.apache.zeppelin.scheduler.Scheduler;
 import org.apache.zeppelin.scheduler.SchedulerFactory;
+import org.apache.zeppelin.spark.utils.SparkSqlCompleter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -292,6 +293,7 @@ public Scheduler getScheduler() {
 
   @Override
   public List<String> completion(String buf, int cursor) {
-    return null;
+    SQLContext sqlc = getSparkInterpreter().getSQLContext();
+    return new SparkSqlCompleter(sqlc).completion(buf, cursor);
   }
 }

spark/src/main/resources/sparksql.txt

@@ -0,0 +1,49 @@
+ALL
+AND
+APPROXIMATE
+AS
+ASC
+BETWEEN
+BY
+CASE
+CAST
+DESC
+DISTINCT
+ELSE
+END
+EXCEPT
+FALSE
+FROM
+FULL
+GROUP
+HAVING
+IN
+INNER
+INSERT
+INTERSECT
+INTO
+IS
+JOIN
+LEFT
+LIKE
+LIMIT
+NOT
+NULL
+ON
+OR
+ORDER
+OUTER
+OVERWRITE
+REGEXP
+RIGHT
+RLIKE
+SELECT
+SEMI
+SORT
+TABLE
+THEN
+TRUE
+UNION
+WHEN
+WHERE
+WITH

spark/src/main/scala/org/apache/zeppelin/spark/utils/SparkSqlCompleter.scala

@@ -0,0 +1,38 @@
+package org.apache.zeppelin.spark.utils
+
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.types.{StructType, StructField}
+
+import scala.collection.JavaConverters._
+import scala.io.Source
+
+class SparkSqlCompleter(val sqlContext: SQLContext) {
+
+  private val keywords = Source.fromInputStream(this.getClass.getResourceAsStream("/sparksql.txt")).getLines().toSet
+
+  def completion(buf: String, cursor: Int): java.util.List[String] = {
+    import sqlContext.implicits._
+
+    val tables = sqlContext.tables().select('tableName).collect().map(_.getString(0))
+    val fields = tables.flatMap(getFields)
+
+    val candidates = (tables ++ fields ++ keywords).toSet
+    val prefix = buf.substring(0, cursor - 1).split("\\s").last.toLowerCase
+    val offset = if (prefix.contains(".")) prefix.lastIndexOf('.') + 1 else 0
+    candidates.filter(_.toLowerCase.startsWith(prefix)).map(_.substring(offset)).toList.asJava
+  }
+
+  private def getFields(table: String): Seq[String] = {
+    def fieldNames(prefix: Seq[String], fields: Array[StructField]): Seq[String] = {
+      fields.flatMap { f =>
+        val sub = f.dataType match {
+          case StructType(subFields) => fieldNames(prefix :+ f.name, subFields)
+          case _ => Seq.empty
+        }
+        sub :+ (prefix :+ f.name).mkString(".")
+      }
+    }
+    fieldNames(Seq.empty, sqlContext.table(table).schema.fields)
+  }
+
+}

zeppelin-web/app/scripts/controllers/paragraph.js

@@ -441,13 +441,14 @@ angular.module('zeppelinWebApp')
           getCompletions : function(editor, session, pos, prefix, callback) {
               if (!$scope.editor.isFocused() ){ return;}
 
-              var buf = session.getTextRange(new Range(0, 0, pos.row, pos.column));
+              var pos = session.getTextRange(new Range(0, 0, pos.row, pos.column)).length;
+              var buf = session.getValue();
               $rootScope.$emit('sendNewEvent', {
                   op : 'COMPLETION',
                   data : {
                       id : $scope.paragraph.id,
                       buf : buf,
-                      cursor : buf.length
+                      cursor : pos
                   }
               });
 
@@ -497,6 +498,8 @@ angular.module('zeppelinWebApp')
       var code = $scope.editor.getSession().getValue();
       if ( String(code).startsWith('%sql')) {
         $scope.editor.getSession().setMode(editorMode.sql);
+      } else if ( String(code).startsWith('%bq')) {
+        $scope.editor.getSession().setMode(editorMode.sql);
       } else if ( String(code).startsWith('%md')) {
         $scope.editor.getSession().setMode(editorMode.markdown);
       } else {