ZEPPELIN-211: Add Geode Interpreter remote completer

geode/pom.xml

@@ -51,6 +51,17 @@
       <version>${geode.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>jline</groupId>
+      <artifactId>jline</artifactId>
+      <version>2.12.1</version>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-exec</artifactId>

geode/src/main/java/org/apache/zeppelin/geode/GeodeOqlInterpreter.java

@@ -14,6 +14,7 @@
  */
 package org.apache.zeppelin.geode;
 
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Properties;
@@ -35,6 +36,8 @@
 import com.gemstone.gemfire.cache.query.SelectResults;
 import com.gemstone.gemfire.cache.query.Struct;
 import com.gemstone.gemfire.pdx.PdxInstance;
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
 
 /**
  * Apache Geode OQL Interpreter (http://geode.incubator.apache.org)
@@ -77,6 +80,9 @@
  *}
  * </p>
  * <p>
+ * Use (Ctrl + .) to activate the auto-completion.
+ * </p>
+ * <p>
  * Known issue:http://gemfire.docs.pivotal.io/bugnotes/KnownIssuesGemFire810.html #43673 Using query
  * "select * from /exampleRegion.entrySet" fails in a client-server topology and/or in a
  * PartitionedRegion.
@@ -99,6 +105,7 @@ public class GeodeOqlInterpreter extends Interpreter {
   public static final String LOCATOR_HOST = "geode.locator.host";
   public static final String LOCATOR_PORT = "geode.locator.port";
   public static final String MAX_RESULT = "geode.max.result";
+  private static final List<String> NO_COMPLETION = new ArrayList<String>();
 
   static {
     Interpreter.register(
@@ -110,11 +117,20 @@ public class GeodeOqlInterpreter extends Interpreter {
             .add(MAX_RESULT, DEFAULT_MAX_RESULT, "Max number of OQL result to display.").build());
   }
 
+  private static final Function<CharSequence, String> sequenceToStringTransformer =
+      new Function<CharSequence, String>() {
+        public String apply(CharSequence seq) {
+          return seq.toString();
+        }
+      };
+
   private ClientCache clientCache = null;
   private QueryService queryService = null;
   private Exception exceptionOnConnect;
   private int maxResult;
 
+  private OqlCompleter oqlCompleter;
+
   public GeodeOqlInterpreter(Properties property) {
     super(property);
   }
@@ -142,6 +158,7 @@ public void open() {
 
       clientCache = getClientCache();
       queryService = clientCache.getQueryService();
+      oqlCompleter = new OqlCompleter(OqlCompleter.getOqlCompleterTokens(clientCache));
 
       exceptionOnConnect = null;
       logger.info("Successfully created Geode connection");
@@ -301,7 +318,12 @@ public Scheduler getScheduler() {
 
   @Override
   public List<String> completion(String buf, int cursor) {
-    return null;
+    List<CharSequence> candidates = new ArrayList<CharSequence>();
+    if (oqlCompleter.complete(buf, cursor, candidates) >= 0) {
+      return Lists.transform(candidates, sequenceToStringTransformer);
+    } else {
+      return NO_COMPLETION;
+    }
   }
 
   public int getMaxResult() {

geode/src/main/java/org/apache/zeppelin/geode/OqlCompleter.java

@@ -0,0 +1,100 @@
+/**
+ * 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.geode;
+
+import static org.apache.commons.lang.StringUtils.isBlank;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.List;
+import java.util.Set;
+import java.util.StringTokenizer;
+import java.util.TreeSet;
+
+import jline.console.completer.ArgumentCompleter.ArgumentList;
+import jline.console.completer.ArgumentCompleter.WhitespaceArgumentDelimiter;
+import jline.console.completer.StringsCompleter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.gemstone.gemfire.cache.client.ClientCache;
+
+/**
+ * OQL auto-completer for the {@link GeodeOqlInterpreter}.
+ */
+public class OqlCompleter extends StringsCompleter {
+
+  private static Logger logger = LoggerFactory.getLogger(OqlCompleter.class);
+
+  private WhitespaceArgumentDelimiter delimiter = new WhitespaceArgumentDelimiter();
+
+  public OqlCompleter(Set<String> completions) {
+    super(completions);
+  }
+
+  @Override
+  public int complete(String buffer, int cursor, List<CharSequence> candidates) {
+
+    if (isBlank(buffer) || cursor > buffer.length() + 1) {
+      return -1;
+    }
+
+    // The delimiter breaks the buffer into separate words (arguments), separated by the
+    // whitespaces.
+    ArgumentList argumentList = delimiter.delimit(buffer, cursor);
+    String argument = argumentList.getCursorArgument();
+    // cursor in the selected argument
+    int argumentPosition = argumentList.getArgumentPosition();
+
+    if (isBlank(argument)) {
+      int argumentsCount = argumentList.getArguments().length;
+      if (argumentsCount <= 0 || ((buffer.length() + 2) < cursor)
+          || delimiter.isDelimiterChar(buffer, cursor - 2)) {
+        return -1;
+      }
+      argument = argumentList.getArguments()[argumentsCount - 1];
+      argumentPosition = argument.length();
+    }
+
+    int complete = super.complete(argument, argumentPosition, candidates);
+
+    logger.debug("complete:" + complete + ", size:" + candidates.size());
+
+    return complete;
+  }
+
+  public static Set<String> getOqlCompleterTokens(ClientCache cache) throws IOException {
+
+    Set<String> completions = new TreeSet<String>();
+
+    // add the default OQL completions
+    String keywords =
+        new BufferedReader(new InputStreamReader(
+            OqlCompleter.class.getResourceAsStream("/oql.keywords"))).readLine();
+
+
+    // Also allow upper-case versions of all the keywords
+    keywords += "," + keywords.toUpperCase();
+
+    StringTokenizer tok = new StringTokenizer(keywords, ",");
+    while (tok.hasMoreTokens()) {
+      completions.add(tok.nextToken());
+    }
+
+    return completions;
+  }
+}

geode/src/main/resources/oql.keywords

@@ -0,0 +1 @@
+all,and,array,as,asc,boolean,by,byte,char,collection,count,date,desc,dictionary,distinct,double,element,false,float,from,import,in,int,is_defined,is_undefined,like,limit,long,map,nil,not,null,nvl,octet,or,order,select,set,short,string,time,timestamp,to_date,true,type,undefined,where,abs,any,andthen,avg,bag,declare,define,enum,except,exists,for,first,flatten,group,having,intersect,interval,last,list,listtoset,max,min,mod,orelse,query,some,struct,sum,undefine,union,unique

geode/src/test/java/org/apache/zeppelin/geode/GeodeOqlInterpreterTest.java

@@ -15,7 +15,6 @@
 package org.apache.zeppelin.geode;
 
 import static org.apache.zeppelin.geode.GeodeOqlInterpreter.*;
-
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
@@ -27,6 +26,7 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
@@ -178,4 +178,20 @@ public void oqlWithExceptionOnConnect() throws Exception {
   public void testFormType() {
     assertEquals(FormType.SIMPLE, new GeodeOqlInterpreter(new Properties()).getFormType());
   }
+
+  @Test
+  public void testAutoCompletion() throws SQLException {
+    Properties properties = new Properties();
+    properties.put(LOCATOR_HOST, DEFAULT_HOST);
+    properties.put(LOCATOR_PORT, DEFAULT_PORT);
+    properties.put(MAX_RESULT, DEFAULT_MAX_RESULT);
+
+    GeodeOqlInterpreter spyGeodeOqlInterpreter = spy(new GeodeOqlInterpreter(properties));
+
+    spyGeodeOqlInterpreter.open();
+
+    assertEquals(1, spyGeodeOqlInterpreter.completion("SEL", 0).size());
+    assertEquals("SELECT ", spyGeodeOqlInterpreter.completion("SEL", 0).iterator().next());
+    assertEquals(0, spyGeodeOqlInterpreter.completion("SEL", 100).size());
+  }
 }

geode/src/test/java/org/apache/zeppelin/geode/OqlCompleterTest.java

@@ -0,0 +1,158 @@
+/**
+ * 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.geode;
+
+import static com.google.common.collect.Sets.newHashSet;
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Set;
+
+import jline.console.completer.Completer;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Joiner;
+
+public class OqlCompleterTest {
+
+  private Logger logger = LoggerFactory.getLogger(OqlCompleterTest.class);
+
+  private final static Set<String> EMPTY = new HashSet<String>();
+
+  private CompleterTester tester;
+
+  @Before
+  public void beforeTest() throws IOException, SQLException {
+    OqlCompleter sqlCompleter = new OqlCompleter(OqlCompleter.getOqlCompleterTokens(null));
+    tester = new CompleterTester(sqlCompleter);
+  }
+
+  @Test
+  public void testEdges() {
+    String buffer = "  ORDER  ";
+    tester.buffer(buffer).from(0).to(8).expect(newHashSet("ORDER ")).test();
+    tester.buffer(buffer).from(9).to(15).expect(EMPTY).test();
+  }
+
+  @Test
+  public void testMultipleWords() {
+    String buffer = "  SELE  fro";
+    tester.buffer(buffer).from(0).to(6).expect(newHashSet("SELECT ")).test();
+    tester.buffer(buffer).from(7).to(12).expect(newHashSet("from ")).test();
+    tester.buffer(buffer).from(13).to(14).expect(EMPTY).test();
+  }
+
+  @Test
+  public void testMultiLineBuffer() {
+    String buffer = " \n SELE \n fro";
+    tester.buffer(buffer).from(0).to(7).expect(newHashSet("SELECT ")).test();
+    tester.buffer(buffer).from(8).to(14).expect(newHashSet("from ")).test();
+    tester.buffer(buffer).from(15).to(16).expect(EMPTY).test();
+  }
+
+  @Test
+  public void testMultipleCompletionSuggestions() {
+    String buffer = "  S";
+    tester.buffer(buffer).from(0).to(4)
+        .expect(newHashSet("STRUCT", "SHORT", "SET", "SUM", "SELECT", "SOME", "STRING")).test();
+    tester.buffer(buffer).from(5).to(7).expect(EMPTY).test();
+  }
+
+  public class CompleterTester {
+
+    private Completer completer;
+
+    private String buffer;
+    private int fromCursor;
+    private int toCursor;
+    private Set<String> expectedCompletions;
+
+    public CompleterTester(Completer completer) {
+      this.completer = completer;
+    }
+
+    public CompleterTester buffer(String buffer) {
+      this.buffer = buffer;
+      return this;
+    }
+
+    public CompleterTester from(int fromCursor) {
+      this.fromCursor = fromCursor;
+      return this;
+    }
+
+    public CompleterTester to(int toCursor) {
+      this.toCursor = toCursor;
+      return this;
+    }
+
+    public CompleterTester expect(Set<String> expectedCompletions) {
+      this.expectedCompletions = expectedCompletions;
+      return this;
+    }
+
+    public void test() {
+      for (int c = fromCursor; c <= toCursor; c++) {
+        expectedCompletions(buffer, c, expectedCompletions);
+      }
+    }
+
+    private void expectedCompletions(String buffer, int cursor, Set<String> expected) {
+
+      ArrayList<CharSequence> candidates = new ArrayList<CharSequence>();
+
+      completer.complete(buffer, cursor, candidates);
+
+      String explain = explain(buffer, cursor, candidates);
+
+      logger.info(explain);
+
+      assertEquals("Buffer [" + buffer.replace(" ", ".") + "] and Cursor[" + cursor + "] "
+          + explain, expected, newHashSet(candidates));
+    }
+
+    private String explain(String buffer, int cursor, ArrayList<CharSequence> candidates) {
+      StringBuffer sb = new StringBuffer();
+
+      for (int i = 0; i <= Math.max(cursor, buffer.length()); i++) {
+        if (i == cursor) {
+          sb.append("(");
+        }
+        if (i >= buffer.length()) {
+          sb.append("_");
+        } else {
+          if (Character.isWhitespace(buffer.charAt(i))) {
+            sb.append(".");
+          } else {
+            sb.append(buffer.charAt(i));
+          }
+        }
+        if (i == cursor) {
+          sb.append(")");
+        }
+      }
+      sb.append(" >> [").append(Joiner.on(",").join(candidates)).append("]");
+
+      return sb.toString();
+    }
+  }
+}