diff --git a/docs/interpreter/jdbc.md b/docs/interpreter/jdbc.md
index b7ac45ae44..aad84ea0c9 100644
--- a/docs/interpreter/jdbc.md
+++ b/docs/interpreter/jdbc.md
@@ -128,6 +128,11 @@ The JDBC interpreter properties are defined by default like below.
     <td></td>
     <td>Сomma separated schema (schema = catalog = database) filters to get metadata for completions. Supports '%' symbol is equivalent to any set of characters. (ex. prod_v_%,public%,info)</td>
   </tr>
+  <tr>
+    <td>default.completer.ttlInSeconds</td>
+    <td>120</td>
+    <td>Time to live sql completer in seconds</td>
+  </tr>
 </table>
 
 If you want to connect other databases such as `Mysql`, `Redshift` and `Hive`, you need to edit the property values.
diff --git a/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCInterpreter.java b/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCInterpreter.java
index 3272bc1370..65d53a4bb7 100644
--- a/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCInterpreter.java
+++ b/jdbc/src/main/java/org/apache/zeppelin/jdbc/JDBCInterpreter.java
@@ -101,6 +101,8 @@ public class JDBCInterpreter extends Interpreter {
   static final String PASSWORD_KEY = "password";
   static final String PRECODE_KEY = "precode";
   static final String COMPLETER_SCHEMA_FILTERS_KEY = "completer.schemaFilters";
+  static final String COMPLETER_TTL_KEY = "completer.ttlInSeconds";
+  static final String DEFAULT_COMPLETER_TTL = "120";
   static final String JDBC_JCEKS_FILE = "jceks.file";
   static final String JDBC_JCEKS_CREDENTIAL_KEY = "jceks.credentialKey";
   static final String PRECODE_KEY_TEMPLATE = "%s.precode";
@@ -128,6 +130,7 @@ public class JDBCInterpreter extends Interpreter {
 
   private final HashMap<String, Properties> basePropretiesMap;
   private final HashMap<String, JDBCUserConfigurations> jdbcUserConfigurationsMap;
+  private final HashMap<String, SqlCompleter> sqlCompletersMap;
 
   private int maxLineResults;
 
@@ -135,6 +138,7 @@ public class JDBCInterpreter extends Interpreter {
     super(property);
     jdbcUserConfigurationsMap = new HashMap<>();
     basePropretiesMap = new HashMap<>();
+    sqlCompletersMap = new HashMap<>();
     maxLineResults = MAX_LINE_DEFAULT;
   }
 
@@ -188,12 +192,20 @@ public class JDBCInterpreter extends Interpreter {
     }
   }
 
-  private SqlCompleter createSqlCompleter(Connection jdbcConnection, String propertyKey) {
+  private SqlCompleter createOrUpdateSqlCompleter(SqlCompleter sqlCompleter, Connection connection,
+      String propertyKey, String buf, int cursor) {
     String schemaFiltersKey = String.format("%s.%s", propertyKey, COMPLETER_SCHEMA_FILTERS_KEY);
-    String filters = getProperty(schemaFiltersKey);
-    SqlCompleter completer = new SqlCompleter();
-    completer.initFromConnection(jdbcConnection, filters);
-    return completer;
+    String sqlCompleterTtlKey = String.format("%s.%s", propertyKey, COMPLETER_TTL_KEY);
+    String schemaFiltersString = getProperty(schemaFiltersKey);
+    int ttlInSeconds = Integer.valueOf(
+        StringUtils.defaultIfEmpty(getProperty(sqlCompleterTtlKey), DEFAULT_COMPLETER_TTL)
+    );
+    if (sqlCompleter == null) {
+      sqlCompleter = new SqlCompleter(ttlInSeconds);
+    }
+
+    sqlCompleter.createOrUpdateFromConnection(connection, schemaFiltersString, buf, cursor);
+    return sqlCompleter;
   }
 
   private void initStatementMap() {
@@ -787,6 +799,10 @@ public class JDBCInterpreter extends Interpreter {
       InterpreterContext interpreterContext) {
     List<InterpreterCompletion> candidates = new ArrayList<>();
     String propertyKey = getPropertyKey(buf);
+    String sqlCompleterKey =
+        String.format("%s.%s", interpreterContext.getAuthenticationInfo().getUser(), propertyKey);
+    SqlCompleter sqlCompleter = sqlCompletersMap.get(sqlCompleterKey);
+
     Connection connection = null;
     try {
       if (interpreterContext != null) {
@@ -796,11 +812,9 @@ public class JDBCInterpreter extends Interpreter {
       logger.warn("SQLCompleter will created without use connection");
     }
 
-    SqlCompleter sqlCompleter = createSqlCompleter(connection, propertyKey);
-
-    if (sqlCompleter != null) {
-      sqlCompleter.complete(buf, cursor - 1, candidates);
-    }
+    sqlCompleter = createOrUpdateSqlCompleter(sqlCompleter, connection, propertyKey, buf, cursor);
+    sqlCompletersMap.put(sqlCompleterKey, sqlCompleter);
+    sqlCompleter.complete(buf, cursor, candidates);
 
     return candidates;
   }
diff --git a/jdbc/src/main/java/org/apache/zeppelin/jdbc/SqlCompleter.java b/jdbc/src/main/java/org/apache/zeppelin/jdbc/SqlCompleter.java
index 704ec597c4..b83b5129be 100644
--- a/jdbc/src/main/java/org/apache/zeppelin/jdbc/SqlCompleter.java
+++ b/jdbc/src/main/java/org/apache/zeppelin/jdbc/SqlCompleter.java
@@ -24,6 +24,7 @@ import java.util.regex.Pattern;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.math.NumberUtils;
+import org.apache.zeppelin.completer.CachedCompleter;
 import org.apache.zeppelin.completer.CompletionType;
 import org.apache.zeppelin.completer.StringsCompleter;
 import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
@@ -33,8 +34,6 @@ import org.slf4j.LoggerFactory;
 import jline.console.completer.ArgumentCompleter.ArgumentList;
 import jline.console.completer.ArgumentCompleter.WhitespaceArgumentDelimiter;
 
-import static org.apache.commons.lang.StringUtils.isBlank;
-
 /**
  * SQL auto complete functionality for the JdbcInterpreter.
  */
@@ -42,6 +41,7 @@ public class SqlCompleter {
 
   private static Logger logger = LoggerFactory.getLogger(SqlCompleter.class);
 
+
   /**
    * Delimiter that can split SQL statement in keyword list
    */
@@ -59,23 +59,30 @@ public class SqlCompleter {
   /**
    * Schema completer
    */
-  private StringsCompleter schemasCompleter = new StringsCompleter();
+  private CachedCompleter schemasCompleter;
 
   /**
    * Contain different completer with table list for every schema name
    */
-  private Map<String, StringsCompleter> tablesCompleters = new HashMap<>();
+  private Map<String, CachedCompleter> tablesCompleters = new HashMap<>();
 
   /**
    * Contains different completer with column list for every table name
    * Table names store as schema_name.table_name
    */
-  private Map<String, StringsCompleter> columnsCompleters = new HashMap<>();
+  private Map<String, CachedCompleter> columnsCompleters = new HashMap<>();
 
   /**
    * Completer for sql keywords
    */
-  private StringsCompleter keywordCompleter = new StringsCompleter();
+  private CachedCompleter keywordCompleter;
+
+  private int ttlInSeconds;
+
+
+  public SqlCompleter(int ttlInSeconds) {
+    this.ttlInSeconds = ttlInSeconds;
+  }
 
   public int complete(String buffer, int cursor, List<InterpreterCompletion> candidates) {
 
@@ -95,25 +102,9 @@ public class SqlCompleter {
       argumentPosition = argumentList.getArgumentPosition();
     }
 
-    boolean isColumnAllowed = true;
-    if (buffer.length() > 0) {
-      String beforeCursorBuffer = buffer.substring(0,
-          Math.min(cursor, buffer.length())).toUpperCase();
-      // check what sql is and where cursor is to allow column completion or not
-      if (beforeCursorBuffer.contains("SELECT ") && beforeCursorBuffer.contains(" FROM ")
-          && !beforeCursorBuffer.contains(" WHERE "))
-        isColumnAllowed = false;
-    }
-
     int complete = completeName(cursorArgument, argumentPosition, candidates,
-            findAliasesInSQL(argumentList.getArguments()), isColumnAllowed);
+            findAliasesInSQL(argumentList.getArguments()));
 
-    if (candidates.size() == 1) {
-      InterpreterCompletion interpreterCompletion = candidates.get(0);
-      interpreterCompletion.setName(interpreterCompletion.getName() + " ");
-      interpreterCompletion.setValue(interpreterCompletion.getValue() + " ");
-      candidates.set(0, interpreterCompletion);
-    }
     logger.debug("complete:" + complete + ", size:" + candidates.size());
     return complete;
   }
@@ -132,6 +123,7 @@ public class SqlCompleter {
     Set<String> res = new HashSet<>();
     try {
       ResultSet schemas = meta.getSchemas();
+
       try {
         while (schemas.next()) {
           String schemaName = schemas.getString("TABLE_SCHEM");
@@ -185,58 +177,40 @@ public class SqlCompleter {
     return res;
   }
 
+
+  private static void fillTableNames(String schema, DatabaseMetaData meta, Set<String> tables) {
+    try (ResultSet tbls = meta.getTables(schema, schema, "%", null)) {
+      while (tbls.next()) {
+        String table = tbls.getString("TABLE_NAME");
+        tables.add(table);
+      }
+    } catch (Throwable t) {
+      logger.error("Failed to retrieve the table name", t);
+    }
+  }
+
   /**
    * Fill two map with list of tables and list of columns
    *
-   * @param catalogName name of a catalog
-   * @param meta metadata from connection to database
-   * @param schemaFilter a schema name pattern; must match the schema name
-   *        as it is stored in the database; "" retrieves those without a schema;
-   *        <code>null</code> means that the schema name should not be used to narrow
-   *        the search; supports '%'; for example "prod_v_%"
-   * @param tables function fills this map, for every schema name adds
-   *        set of table names within the schema
-   * @param columns function fills this map, for every table name adds set
+   * @param schema name of a scheme
+   * @param table name of a table
+   * @param meta meta metadata from connection to database
+   * @param columns function fills this set, for every table name adds set
    *        of columns within the table; table name is in format schema_name.table_name
    */
-  private static void fillTableAndColumnNames(String catalogName, DatabaseMetaData meta,
-                                              String schemaFilter,
-                                              Map<String, Set<String>> tables,
-                                              Map<String, Set<String>> columns)  {
-    try {
-      ResultSet cols = meta.getColumns(catalogName, StringUtils.EMPTY, "%", "%");
-      try {
-        while (cols.next()) {
-          String schema = cols.getString("TABLE_SCHEM");
-          if (schema == null) {
-            schema = cols.getString("TABLE_CAT");
-          }
-          if (!schemaFilter.equals("") && !schema.matches(schemaFilter.replace("%", ".*?"))) {
-            continue;
-          }
-          String table = cols.getString("TABLE_NAME");
-          String column = cols.getString("COLUMN_NAME");
-          if (!isBlank(table)) {
-            String schemaTable = schema + "." + table;
-            if (!columns.containsKey(schemaTable)) {
-              columns.put(schemaTable, new HashSet<String>());
-            }
-            columns.get(schemaTable).add(column);
-            if (!tables.containsKey(schema)) {
-              tables.put(schema, new HashSet<String>());
-            }
-            tables.get(schema).add(table);
-          }
-        }
-      } finally {
-        cols.close();
+  private static void fillColumnNames(String schema, String table, DatabaseMetaData meta,
+      Set<String> columns) {
+    try (ResultSet cols = meta.getColumns(schema, schema, table, "%")) {
+      while (cols.next()) {
+        String column = cols.getString("COLUMN_NAME");
+        columns.add(column);
       }
     } catch (Throwable t) {
       logger.error("Failed to retrieve the column name", t);
     }
   }
 
-  public static Set<String> getSqlKeywordsCompletions(Connection connection) throws IOException,
+  public static Set<String> getSqlKeywordsCompletions(DatabaseMetaData meta) throws IOException,
           SQLException {
 
     // Add the default SQL completions
@@ -246,12 +220,11 @@ public class SqlCompleter {
 
     Set<String> completions = new TreeSet<>();
 
-    if (null != connection) {
-      DatabaseMetaData metaData = connection.getMetaData();
+    if (null != meta) {
 
       // Add the driver specific SQL completions
       String driverSpecificKeywords =
-              "/" + metaData.getDriverName().replace(" ", "-").toLowerCase() + "-sql.keywords";
+              "/" + meta.getDriverName().replace(" ", "-").toLowerCase() + "-sql.keywords";
       logger.info("JDBC DriverName:" + driverSpecificKeywords);
       try {
         if (SqlCompleter.class.getResource(driverSpecificKeywords) != null) {
@@ -269,27 +242,27 @@ public class SqlCompleter {
 
       // Add the keywords from the current JDBC connection
       try {
-        keywords += "," + metaData.getSQLKeywords();
+        keywords += "," + meta.getSQLKeywords();
       } catch (Exception e) {
         logger.debug("fail to get SQL key words from database metadata: " + e, e);
       }
       try {
-        keywords += "," + metaData.getStringFunctions();
+        keywords += "," + meta.getStringFunctions();
       } catch (Exception e) {
         logger.debug("fail to get string function names from database metadata: " + e, e);
       }
       try {
-        keywords += "," + metaData.getNumericFunctions();
+        keywords += "," + meta.getNumericFunctions();
       } catch (Exception e) {
         logger.debug("fail to get numeric function names from database metadata: " + e, e);
       }
       try {
-        keywords += "," + metaData.getSystemFunctions();
+        keywords += "," + meta.getSystemFunctions();
       } catch (Exception e) {
         logger.debug("fail to get system function names from database metadata: " + e, e);
       }
       try {
-        keywords += "," + metaData.getTimeDateFunctions();
+        keywords += "," + meta.getTimeDateFunctions();
       } catch (Exception e) {
         logger.debug("fail to get time date function names from database metadata: " + e, e);
       }
@@ -307,95 +280,80 @@ public class SqlCompleter {
     return completions;
   }
 
-  /**
-   * Initializes local schema completers from list of schema names
-   *
-   * @param schemas set of schema names
-   */
-  private void initSchemas(Set<String> schemas) {
-    schemasCompleter = new StringsCompleter(new TreeSet<>(schemas));
-  }
-
-  /**
-   * Initializes local table completers from list of table name
-   *
-   * @param tables for every schema name there is a set of table names within the schema
-   */
-  private void initTables(Map<String, Set<String>> tables) {
-    tablesCompleters.clear();
-    for (Map.Entry<String, Set<String>> entry : tables.entrySet()) {
-      tablesCompleters.put(entry.getKey(), new StringsCompleter(new TreeSet<>(entry.getValue())));
-    }
-  }
-
-  /**
-   * Initializes local column completers from list of column names
-   *
-   * @param columns for every table name there is a set of columns within the table;
-   *        table name is in format schema_name.table_name
-   */
-  private void initColumns(Map<String, Set<String>> columns) {
-    columnsCompleters.clear();
-    for (Map.Entry<String, Set<String>> entry : columns.entrySet()) {
-      columnsCompleters.put(entry.getKey(), new StringsCompleter(new TreeSet<>(entry.getValue())));
-    }
-  }
-
-  /**
-   * Initializes all local completers
-   *
-   * @param schemas set of schema names
-   * @param tables for every schema name there is a set of table names within the schema
-   * @param columns for every table name there is a set of columns within the table;
-   *        table name is in format schema_name.table_name
-   * @param keywords set with sql keywords
-   */
-  public void init(Set<String> schemas, Map<String, Set<String>> tables,
-                   Map<String, Set<String>> columns, Set<String> keywords) {
-    initSchemas(schemas);
-    initTables(tables);
-    initColumns(columns);
-    keywordCompleter = new StringsCompleter(keywords);
-  }
-
   /**
    * Initializes all local completers from database connection
    *
    * @param connection database connection
-   * @param schemaFiltersString a comma separated schema name patterns; supports '%'  symbol;
-   * for example "prod_v_%,prod_t_%"
+   * @param schemaFiltersString a comma separated schema name patterns, supports '%'  symbol;
+   *        for example "prod_v_%,prod_t_%"
    */
-  public void initFromConnection(Connection connection, String schemaFiltersString) {
+  public void createOrUpdateFromConnection(Connection connection, String schemaFiltersString,
+      String buffer, int cursor) {
     if (schemaFiltersString == null) {
       schemaFiltersString = StringUtils.EMPTY;
     }
     List<String> schemaFilters = Arrays.asList(schemaFiltersString.split(","));
+    CursorArgument cursorArgument = parseCursorArgument(buffer, cursor);
 
     try (Connection c = connection) {
-      Map<String, Set<String>> tables = new HashMap<>();
-      Map<String, Set<String>> columns = new HashMap<>();
+      Set<String> tables = new HashSet<>();
+      Set<String> columns = new HashSet<>();
       Set<String> schemas = new HashSet<>();
       Set<String> catalogs = new HashSet<>();
-      Set<String> keywords = getSqlKeywordsCompletions(connection);
-      if (connection != null) {
-        schemas = getSchemaNames(connection.getMetaData(), schemaFilters);
-        catalogs = getCatalogNames(connection.getMetaData(), schemaFilters);
-        if (schemas.size() == 0) {
-          schemas.addAll(catalogs);
+      Set<String> keywords = new HashSet<>();
+
+      if (c != null) {
+        DatabaseMetaData databaseMetaData = c.getMetaData();
+        if (keywordCompleter == null || keywordCompleter.getCompleter() == null
+            || keywordCompleter.isExpired()) {
+          keywords = getSqlKeywordsCompletions(databaseMetaData);
+          keywordCompleter = new CachedCompleter(new StringsCompleter(keywords), 0);
         }
-        for (String schema : schemas) {
-          for (String schemaFilter : schemaFilters) {
-            fillTableAndColumnNames(schema, connection.getMetaData(), schemaFilter, tables,
-                columns);
+        if (cursorArgument.needLoadSchemas() &&
+            (schemasCompleter == null || schemasCompleter.getCompleter() == null
+            || schemasCompleter.isExpired())) {
+          schemas = getSchemaNames(databaseMetaData, schemaFilters);
+          catalogs = getCatalogNames(databaseMetaData, schemaFilters);
+
+          if (schemas.size() == 0) {
+            schemas.addAll(catalogs);
+          }
+          if (!schemas.isEmpty()) {
+            schemasCompleter = new CachedCompleter(
+                new StringsCompleter(new TreeSet<>(schemas)), ttlInSeconds);
           }
         }
+
+        CachedCompleter tablesCompleter = tablesCompleters.get(cursorArgument.getSchema());
+        if (cursorArgument.needLoadTables() &&
+            (tablesCompleter == null || tablesCompleter.isExpired())) {
+          fillTableNames(cursorArgument.getSchema(), databaseMetaData, tables);
+          if (!tables.isEmpty()) {
+            tablesCompleters.put(cursorArgument.getSchema(), new CachedCompleter(
+                new StringsCompleter(new TreeSet<>(tables)), ttlInSeconds));
+          }
+        }
+
+        String schemaTable =
+            String.format("%s.%s", cursorArgument.getSchema(), cursorArgument.getTable());
+        CachedCompleter columnsCompleter = columnsCompleters.get(schemaTable);
+
+        if (cursorArgument.needLoadColumns() &&
+            (columnsCompleter == null || columnsCompleter.isExpired())) {
+          fillColumnNames(cursorArgument.getSchema(), cursorArgument.getTable(), databaseMetaData,
+              columns);
+          if (!columns.isEmpty()) {
+            columnsCompleters.put(schemaTable,
+                new CachedCompleter(new StringsCompleter(columns), ttlInSeconds));
+          }
+        }
+
+        logger.info("Completer initialized with " + schemas.size() + " schemas, " +
+            columns.size() + " tables and " + keywords.size() + " keywords");
       }
-      init(schemas, tables, columns, keywords);
-      logger.info("Completer initialized with " + schemas.size() + " schemas, " +
-              columns.size() + " tables and " + keywords.size() + " keywords");
 
     } catch (SQLException | IOException e) {
-      logger.error("Failed to update the metadata conmpletions", e);
+      logger.error("Failed to update the metadata completions", e);
     }
   }
 
@@ -422,7 +380,7 @@ public class SqlCompleter {
    * @return -1 in case of no candidates found, 0 otherwise
    */
   private int completeKeyword(String buffer, int cursor, List<CharSequence> candidates) {
-    return keywordCompleter.complete(buffer, cursor, candidates);
+    return keywordCompleter.getCompleter().complete(buffer, cursor, candidates);
   }
 
   /**
@@ -431,7 +389,7 @@ public class SqlCompleter {
    * @return -1 in case of no candidates found, 0 otherwise
    */
   private int completeSchema(String buffer, int cursor, List<CharSequence> candidates) {
-    return schemasCompleter.complete(buffer, cursor, candidates);
+    return schemasCompleter.getCompleter().complete(buffer, cursor, candidates);
   }
 
   /**
@@ -441,21 +399,12 @@ public class SqlCompleter {
    */
   private int completeTable(String schema, String buffer, int cursor,
                             List<CharSequence> candidates) {
-    if (schema == null) {
-      int res = -1;
-      Set<CharSequence> candidatesSet = new HashSet<>();
-      for (StringsCompleter stringsCompleter : tablesCompleters.values()) {
-        int resTable = stringsCompleter.complete(buffer, cursor, candidatesSet);
-        res = Math.max(res, resTable);
-      }
-      candidates.addAll(candidatesSet);
-      return res;
-    }
     // Wrong schema
-    if (!tablesCompleters.containsKey(schema) && schema != null)
+    if (schema == null || !tablesCompleters.containsKey(schema))
       return -1;
-    else
-      return tablesCompleters.get(schema).complete(buffer, cursor, candidates);
+    else {
+      return tablesCompleters.get(schema).getCompleter().complete(buffer, cursor, candidates);
+    }
   }
 
   /**
@@ -465,22 +414,12 @@ public class SqlCompleter {
    */
   private int completeColumn(String schema, String table, String buffer, int cursor,
                              List<CharSequence> candidates) {
-    if (table == null && schema == null) {
-      int res = -1;
-      Set<CharSequence> candidatesSet = new HashSet<>();
-      for (StringsCompleter stringsCompleter : columnsCompleters.values()) {
-        int resColumn = stringsCompleter.complete(buffer, cursor, candidatesSet);
-        res = Math.max(res, resColumn);
-      }
-      candidates.addAll(candidatesSet);
-      return res;
-    }
     // Wrong schema or wrong table
-    if (!tablesCompleters.containsKey(schema) ||
-        !columnsCompleters.containsKey(schema + "." + table)) {
+    if (schema == null || table == null || !columnsCompleters.containsKey(schema + "." + table)) {
       return -1;
     } else {
-      return columnsCompleters.get(schema + "." + table).complete(buffer, cursor, candidates);
+      return columnsCompleters.get(schema + "." + table).getCompleter()
+          .complete(buffer, cursor, candidates);
     }
   }
 
@@ -489,74 +428,56 @@ public class SqlCompleter {
    * a schema, a table of a column or a keyword
    *
    * @param aliases for every alias contains table name in format schema_name.table_name
-   * @param isColumnAllowed if false the function will not search and complete columns
    * @return -1 in case of no candidates found, 0 otherwise
    */
   public int completeName(String buffer, int cursor, List<InterpreterCompletion> candidates,
-                          Map<String, String> aliases, boolean isColumnAllowed) {
+                          Map<String, String> aliases) {
 
-    // points divide the name to the schema, table and column - find them
-    int pointPos1 = -1;
-    int pointPos2 = -1;
+    CursorArgument cursorArgument = parseCursorArgument(buffer, cursor);
 
-    if (StringUtils.isNotEmpty(buffer)) {
-      if (buffer.length() > cursor) {
-        buffer = buffer.substring(0, cursor + 1);
-      }
-      pointPos1 = buffer.indexOf('.');
-      pointPos2 = buffer.indexOf('.', pointPos1 + 1);
-    }
     // find schema and table name if they are
     String schema;
     String table;
     String column;
 
-    if (pointPos1 == -1) {             // process all
+    if (cursorArgument.getSchema() == null) {             // process all
       List<CharSequence> keywordsCandidates = new ArrayList();
       List<CharSequence> schemaCandidates = new ArrayList<>();
-      List<CharSequence> tableCandidates = new ArrayList<>();
-      List<CharSequence> columnCandidates = new ArrayList<>();
       int keywordsRes = completeKeyword(buffer, cursor, keywordsCandidates);
       int schemaRes = completeSchema(buffer, cursor, schemaCandidates);
-      int tableRes = completeTable(null, buffer, cursor, tableCandidates);
-      int columnRes = -1;
-      if (isColumnAllowed) {
-        columnRes = completeColumn(null, null, buffer, cursor, columnCandidates);
-      }
       addCompletions(candidates, keywordsCandidates, CompletionType.keyword.name());
       addCompletions(candidates, schemaCandidates, CompletionType.schema.name());
-      addCompletions(candidates, tableCandidates, CompletionType.table.name());
-      addCompletions(candidates, columnCandidates, CompletionType.column.name());
-      return NumberUtils.max(new int[]{keywordsRes, schemaRes, tableRes, columnRes});
+      return NumberUtils.max(new int[]{keywordsRes, schemaRes});
     } else {
-      schema = buffer.substring(0, pointPos1);
+      schema = cursorArgument.getSchema();
       if (aliases.containsKey(schema)) {  // process alias case
         String alias = aliases.get(schema);
         int pointPos = alias.indexOf('.');
         schema = alias.substring(0, pointPos);
         table = alias.substring(pointPos + 1);
-        column = buffer.substring(pointPos1 + 1);
-      } else if (pointPos2 == -1) {        // process schema.table case
+        column = cursorArgument.getColumn();
+        List<CharSequence> columnCandidates = new ArrayList();
+        int columnRes = completeColumn(schema, table, column, cursorArgument.getCursorPosition(),
+            columnCandidates);
+        addCompletions(candidates, columnCandidates, CompletionType.column.name());
+        // process schema.table case
+      } else if (cursorArgument.getTable() != null && cursorArgument.getColumn() == null) {
         List<CharSequence> tableCandidates = new ArrayList();
-        table = buffer.substring(pointPos1 + 1);
-        int tableRes = completeTable(schema, table, cursor - pointPos1 - 1, tableCandidates);
+        table = cursorArgument.getTable();
+        int tableRes = completeTable(schema, table, cursorArgument.getCursorPosition(),
+            tableCandidates);
         addCompletions(candidates, tableCandidates, CompletionType.table.name());
         return tableRes;
       } else {
-        table = buffer.substring(pointPos1 + 1, pointPos2);
-        column = buffer.substring(pointPos2 + 1);
+        List<CharSequence> columnCandidates = new ArrayList();
+        table = cursorArgument.getTable();
+        column = cursorArgument.getColumn();
+        int columnRes = completeColumn(schema, table, column, cursorArgument.getCursorPosition(),
+            columnCandidates);
+        addCompletions(candidates, columnCandidates, CompletionType.column.name());
       }
     }
 
-    // here in case of column
-    if (table != null && isColumnAllowed) {
-      List<CharSequence> columnCandidates = new ArrayList();
-      int columnRes = completeColumn(schema, table, column, cursor - pointPos2 - 1,
-          columnCandidates);
-      addCompletions(candidates, columnCandidates, CompletionType.column.name());
-      return columnRes;
-    }
-
     return -1;
   }
 
@@ -572,4 +493,90 @@ public class SqlCompleter {
           candidate.toString(), meta));
     }
   }
+
+  private CursorArgument parseCursorArgument(String buffer, int cursor) {
+    CursorArgument result = new CursorArgument();
+    String buf = buffer.substring(0, cursor);
+    if (StringUtils.isNotEmpty(buf)) {
+      ArgumentList argumentList = sqlDelimiter.delimit(buf, cursor);
+      String cursorArgument = argumentList.getCursorArgument();
+      if (cursorArgument != null) {
+        int pointPos1 = cursorArgument.indexOf('.');
+        int pointPos2 = cursorArgument.indexOf('.', pointPos1 + 1);
+        if (pointPos1 > -1) {
+          result.setSchema(cursorArgument.substring(0, pointPos1).trim());
+          if (pointPos2 > -1) {
+            result.setTable(cursorArgument.substring(pointPos1 + 1, pointPos2));
+            result.setColumn(cursorArgument.substring(pointPos2 + 1));
+            result.setCursorPosition(cursor - pointPos2 - 1);
+          } else {
+            result.setTable(cursorArgument.substring(pointPos1 + 1));
+            result.setCursorPosition(cursor - pointPos1 - 1);
+          }
+        }
+      }
+    }
+
+    return result;
+  }
+
+  private class CursorArgument {
+    private String schema;
+    private String table;
+    private String column;
+    private int cursorPosition;
+
+    public String getSchema() {
+      return schema;
+    }
+
+    public void setSchema(String schema) {
+      this.schema = schema;
+    }
+
+    public String getTable() {
+      return table;
+    }
+
+    public void setTable(String table) {
+      this.table = table;
+    }
+
+    public String getColumn() {
+      return column;
+    }
+
+    public void setColumn(String column) {
+      this.column = column;
+    }
+
+    public int getCursorPosition() {
+      return cursorPosition;
+    }
+
+    public void setCursorPosition(int cursorPosition) {
+      this.cursorPosition = cursorPosition;
+    }
+
+    public boolean needLoadSchemas() {
+      if (table == null && column == null) {
+        return true;
+      }
+      return false;
+    }
+
+    public boolean needLoadTables() {
+      if (schema != null && table != null && column == null) {
+        return true;
+      }
+      return false;
+    }
+
+    public boolean needLoadColumns() {
+      if (schema != null && table != null && column != null) {
+        return true;
+      }
+      return false;
+    }
+  }
 }
diff --git a/jdbc/src/main/resources/interpreter-setting.json b/jdbc/src/main/resources/interpreter-setting.json
index fb8b8b2a23..b1d8c232b5 100644
--- a/jdbc/src/main/resources/interpreter-setting.json
+++ b/jdbc/src/main/resources/interpreter-setting.json
@@ -22,6 +22,12 @@
         "defaultValue": "",
         "description": "The JDBC user password"
       },
+      "default.completer.ttlInSeconds": {
+        "envName": null,
+        "propertyName": "default.completer.ttlInSeconds",
+        "defaultValue": "120",
+        "description": "Time to live sql completer in seconds"
+      },
       "default.driver": {
         "envName": null,
         "propertyName": "default.driver",
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/completer/CachedCompleter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/completer/CachedCompleter.java
new file mode 100644
index 0000000000..ef2223eb2c
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/completer/CachedCompleter.java
@@ -0,0 +1,44 @@
+/**
+ * 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.completer;
+
+import jline.console.completer.Completer;
+
+/**
+ * Completer with time to live
+ */
+public class CachedCompleter {
+  private Completer completer;
+  private int ttlInSeconds;
+  private long createdAt;
+
+  public CachedCompleter(Completer completer, int ttlInSeconds) {
+    this.completer = completer;
+    this.ttlInSeconds = ttlInSeconds;
+    this.createdAt = System.currentTimeMillis();
+  }
+
+  public boolean isExpired() {
+    if (ttlInSeconds == -1 || (ttlInSeconds > 0 &&
+        (System.currentTimeMillis() - createdAt) / 1000 > ttlInSeconds)) {
+      return true;
+    }
+    return false;
+  }
+
+  public Completer getCompleter() {
+    return completer;
+  }
+}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/completer/StringsCompleter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/completer/StringsCompleter.java
index c7dcebe79f..c117441058 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/completer/StringsCompleter.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/completer/StringsCompleter.java
@@ -61,8 +61,9 @@ public class StringsCompleter implements Completer {
     if (buffer == null) {
       candidates.addAll(strings);
     } else {
-      String bufferTmp = buffer.toUpperCase();
-      for (String match : strings.tailSet(buffer)) {
+      String part = buffer.substring(0, cursor);
+      String bufferTmp = part.toUpperCase();
+      for (String match : strings.tailSet(part)) {
         String matchTmp = match.toUpperCase();
         if (!matchTmp.startsWith(bufferTmp)) {
           break;